Repository: dask/dask

Issue:
blocksize option to read_csv doesn't accept floats
The read_csv docstring presents floats as supported:
```
2.  In some cases it can break up large files as follows:

    >>> df = dd.read_csv('largefile.csv', blocksize=25e6)  # 25MB chunks  # doctest: +SKIP
```

but on Python 3 at least you get an error:
```
  File "/home/antoine/miniconda3/envs/cfmdemo/lib/python3.6/site-packages/dask/bytes/core.py", line 166, in read_bytes
    off = list(range(0, size, blocksize))
TypeError: 'float' object cannot be interpreted as an integer
```

Discussion:

Patch:
diff --git a/dask/array/ufunc.py b/dask/array/ufunc.py
index 7096c4071..e8e21f22c 100644
--- a/dask/array/ufunc.py
+++ b/dask/array/ufunc.py
@@ -1,10 +1,13 @@
 from __future__ import absolute_import, division, print_function
 
 from operator import getitem
+from functools import partial
 
 import numpy as np
+from toolz import curry
 
-from .core import Array, elemwise
+from .core import Array, elemwise, atop, apply_infer_dtype, asarray
+from ..base import Base
 from .. import core, sharedict
 from ..utils import skip_doctest
 
@@ -13,6 +16,12 @@ def __array_wrap__(numpy_ufunc, x, *args, **kwargs):
     return x.__array_wrap__(numpy_ufunc(x, *args, **kwargs))
 
 
+@curry
+def copy_docstring(target, source=None):
+    target.__doc__ = skip_doctest(source.__doc__)
+    return target
+
+
 def wrap_elemwise(numpy_ufunc, array_wrap=False):
     """ Wrap up numpy function into dask.array """
 
@@ -33,112 +42,183 @@ def wrap_elemwise(numpy_ufunc, array_wrap=False):
     return wrapped
 
 
+class ufunc(object):
+    _forward_attrs = {'nin', 'nargs', 'nout', 'ntypes', 'identity',
+                      'signature', 'types'}
+
+    def __init__(self, ufunc):
+        if not isinstance(ufunc, np.ufunc):
+            raise TypeError("must be an instance of `ufunc`, "
+                            "got `%s" % type(ufunc).__name__)
+        self._ufunc = ufunc
+        self.__name__ = ufunc.__name__
+        copy_docstring(self, ufunc)
+
+    def __getattr__(self, key):
+        if key in self._forward_attrs:
+            return getattr(self._ufunc, key)
+        raise AttributeError("%r object has no attribute "
+                             "%r" % (type(self).__name__, key))
+
+    def __dir__(self):
+        return list(self._forward_attrs.union(dir(type(self)), self.__dict__))
+
+    def __repr__(self):
+        return repr(self._ufunc)
+
+    def __call__(self, *args, **kwargs):
+        dsk = [arg for arg in args if hasattr(arg, '_elemwise')]
+        if len(dsk) > 0:
+            return dsk[0]._elemwise(self._ufunc, *args, **kwargs)
+        else:
+            return self._ufunc(*args, **kwargs)
+
+    @copy_docstring(source=np.ufunc.outer)
+    def outer(self, A, B, **kwargs):
+        if self.nin != 2:
+            raise ValueError("outer product only supported for binary functions")
+        if 'out' in kwargs:
+            raise ValueError("`out` kwarg not supported")
+
+        A_is_dask = isinstance(A, Base)
+        B_is_dask = isinstance(B, Base)
+        if not A_is_dask and not B_is_dask:
+            return self._ufunc.outer(A, B, **kwargs)
+        elif (A_is_dask and not isinstance(A, Array) or
+              B_is_dask and not isinstance(B, Array)):
+            raise NotImplementedError("Dask objects besides `dask.array.Array` "
+                                      "are not supported at this time.")
+
+        A = asarray(A)
+        B = asarray(B)
+        ndim = A.ndim + B.ndim
+        out_inds = tuple(range(ndim))
+        A_inds = out_inds[:A.ndim]
+        B_inds = out_inds[A.ndim:]
+
+        dtype = apply_infer_dtype(self._ufunc.outer, [A, B], kwargs,
+                                  'ufunc.outer', suggest_dtype=False)
+
+        if 'dtype' in kwargs:
+            func = partial(self._ufunc.outer, dtype=kwargs.pop('dtype'))
+        else:
+            func = self._ufunc.outer
+
+        return atop(func, out_inds, A, A_inds, B, B_inds, dtype=dtype,
+                    token=self.__name__ + '.outer', **kwargs)
+
+
 # ufuncs, copied from this page:
 # http://docs.scipy.org/doc/numpy/reference/ufuncs.html
 
 # math operations
-add = wrap_elemwise(np.add)
-subtract = wrap_elemwise(np.subtract)
-multiply = wrap_elemwise(np.multiply)
-divide = wrap_elemwise(np.divide)
-logaddexp = wrap_elemwise(np.logaddexp)
-logaddexp2 = wrap_elemwise(np.logaddexp2)
-true_divide = wrap_elemwise(np.true_divide)
-floor_divide = wrap_elemwise(np.floor_divide)
-negative = wrap_elemwise(np.negative)
-power = wrap_elemwise(np.power)
-remainder = wrap_elemwise(np.remainder)
-mod = wrap_elemwise(np.mod)
+add = ufunc(np.add)
+subtract = ufunc(np.subtract)
+multiply = ufunc(np.multiply)
+divide = ufunc(np.divide)
+logaddexp = ufunc(np.logaddexp)
+logaddexp2 = ufunc(np.logaddexp2)
+true_divide = ufunc(np.true_divide)
+floor_divide = ufunc(np.floor_divide)
+negative = ufunc(np.negative)
+power = ufunc(np.power)
+remainder = ufunc(np.remainder)
+mod = ufunc(np.mod)
 # fmod: see below
-conj = wrap_elemwise(np.conj)
-exp = wrap_elemwise(np.exp)
-exp2 = wrap_elemwise(np.exp2)
-log = wrap_elemwise(np.log)
-log2 = wrap_elemwise(np.log2)
-log10 = wrap_elemwise(np.log10)
-log1p = wrap_elemwise(np.log1p)
-expm1 = wrap_elemwise(np.expm1)
-sqrt = wrap_elemwise(np.sqrt)
-square = wrap_elemwise(np.square)
-cbrt = wrap_elemwise(np.cbrt)
-reciprocal = wrap_elemwise(np.reciprocal)
+conj = ufunc(np.conj)
+exp = ufunc(np.exp)
+exp2 = ufunc(np.exp2)
+log = ufunc(np.log)
+log2 = ufunc(np.log2)
+log10 = ufunc(np.log10)
+log1p = ufunc(np.log1p)
+expm1 = ufunc(np.expm1)
+sqrt = ufunc(np.sqrt)
+square = ufunc(np.square)
+cbrt = ufunc(np.cbrt)
+reciprocal = ufunc(np.reciprocal)
 
 # trigonometric functions
-sin = wrap_elemwise(np.sin)
-cos = wrap_elemwise(np.cos)
-tan = wrap_elemwise(np.tan)
-arcsin = wrap_elemwise(np.arcsin)
-arccos = wrap_elemwise(np.arccos)
-arctan = wrap_elemwise(np.arctan)
-arctan2 = wrap_elemwise(np.arctan2)
-hypot = wrap_elemwise(np.hypot)
-sinh = wrap_elemwise(np.sinh)
-cosh = wrap_elemwise(np.cosh)
-tanh = wrap_elemwise(np.tanh)
-arcsinh = wrap_elemwise(np.arcsinh)
-arccosh = wrap_elemwise(np.arccosh)
-arctanh = wrap_elemwise(np.arctanh)
-deg2rad = wrap_elemwise(np.deg2rad)
-rad2deg = wrap_elemwise(np.rad2deg)
+sin = ufunc(np.sin)
+cos = ufunc(np.cos)
+tan = ufunc(np.tan)
+arcsin = ufunc(np.arcsin)
+arccos = ufunc(np.arccos)
+arctan = ufunc(np.arctan)
+arctan2 = ufunc(np.arctan2)
+hypot = ufunc(np.hypot)
+sinh = ufunc(np.sinh)
+cosh = ufunc(np.cosh)
+tanh = ufunc(np.tanh)
+arcsinh = ufunc(np.arcsinh)
+arccosh = ufunc(np.arccosh)
+arctanh = ufunc(np.arctanh)
+deg2rad = ufunc(np.deg2rad)
+rad2deg = ufunc(np.rad2deg)
 
 # comparison functions
-greater = wrap_elemwise(np.greater)
-greater_equal = wrap_elemwise(np.greater_equal)
-less = wrap_elemwise(np.less)
-less_equal = wrap_elemwise(np.less_equal)
-not_equal = wrap_elemwise(np.not_equal)
-equal = wrap_elemwise(np.equal)
-logical_and = wrap_elemwise(np.logical_and)
-logical_or = wrap_elemwise(np.logical_or)
-logical_xor = wrap_elemwise(np.logical_xor)
-logical_not = wrap_elemwise(np.logical_not)
-maximum = wrap_elemwise(np.maximum)
-minimum = wrap_elemwise(np.minimum)
-fmax = wrap_elemwise(np.fmax)
-fmin = wrap_elemwise(np.fmin)
+greater = ufunc(np.greater)
+greater_equal = ufunc(np.greater_equal)
+less = ufunc(np.less)
+less_equal = ufunc(np.less_equal)
+not_equal = ufunc(np.not_equal)
+equal = ufunc(np.equal)
+logical_and = ufunc(np.logical_and)
+logical_or = ufunc(np.logical_or)
+logical_xor = ufunc(np.logical_xor)
+logical_not = ufunc(np.logical_not)
+maximum = ufunc(np.maximum)
+minimum = ufunc(np.minimum)
+fmax = ufunc(np.fmax)
+fmin = ufunc(np.fmin)
 
 # floating functions
-isreal = wrap_elemwise(np.isreal, array_wrap=True)
-iscomplex = wrap_elemwise(np.iscomplex, array_wrap=True)
-isfinite = wrap_elemwise(np.isfinite)
-isinf = wrap_elemwise(np.isinf)
-isnan = wrap_elemwise(np.isnan)
-signbit = wrap_elemwise(np.signbit)
-copysign = wrap_elemwise(np.copysign)
-nextafter = wrap_elemwise(np.nextafter)
-spacing = wrap_elemwise(np.spacing)
+isfinite = ufunc(np.isfinite)
+isinf = ufunc(np.isinf)
+isnan = ufunc(np.isnan)
+signbit = ufunc(np.signbit)
+copysign = ufunc(np.copysign)
+nextafter = ufunc(np.nextafter)
+spacing = ufunc(np.spacing)
 # modf: see below
-ldexp = wrap_elemwise(np.ldexp)
+ldexp = ufunc(np.ldexp)
 # frexp: see below
-fmod = wrap_elemwise(np.fmod)
-floor = wrap_elemwise(np.floor)
-ceil = wrap_elemwise(np.ceil)
-trunc = wrap_elemwise(np.trunc)
-divide = wrap_elemwise(np.divide)
+fmod = ufunc(np.fmod)
+floor = ufunc(np.floor)
+ceil = ufunc(np.ceil)
+trunc = ufunc(np.trunc)
+divide = ufunc(np.divide)
 
 # more math routines, from this page:
 # http://docs.scipy.org/doc/numpy/reference/routines.math.html
-degrees = wrap_elemwise(np.degrees)
-radians = wrap_elemwise(np.radians)
-
-rint = wrap_elemwise(np.rint)
-fix = wrap_elemwise(np.fix, array_wrap=True)
-
-angle = wrap_elemwise(np.angle, array_wrap=True)
+degrees = ufunc(np.degrees)
+radians = ufunc(np.radians)
+rint = ufunc(np.rint)
+fabs = ufunc(np.fabs)
+sign = ufunc(np.sign)
+absolute = ufunc(np.absolute)
+
+# non-ufunc elementwise functions
+clip = wrap_elemwise(np.clip)
+isreal = wrap_elemwise(np.isreal, array_wrap=True)
+iscomplex = wrap_elemwise(np.iscomplex, array_wrap=True)
 real = wrap_elemwise(np.real, array_wrap=True)
 imag = wrap_elemwise(np.imag, array_wrap=True)
+fix = wrap_elemwise(np.fix, array_wrap=True)
+i0 = wrap_elemwise(np.i0, array_wrap=True)
+sinc = wrap_elemwise(np.sinc, array_wrap=True)
+nan_to_num = wrap_elemwise(np.nan_to_num, array_wrap=True)
 
-clip = wrap_elemwise(np.clip)
-fabs = wrap_elemwise(np.fabs)
-sign = wrap_elemwise(np.sign)
-absolute = wrap_elemwise(np.absolute)
-
-i0 = wrap_elemwise(np.i0)
-sinc = wrap_elemwise(np.sinc)
 
-nan_to_num = wrap_elemwise(np.nan_to_num)
+@copy_docstring(source=np.angle)
+def angle(x, deg=0):
+    deg = bool(deg)
+    if hasattr(x, '_elemwise'):
+        return x._elemwise(__array_wrap__, np.angle, x, deg)
+    return np.angle(x, deg=deg)
 
 
+@copy_docstring(source=np.frexp)
 def frexp(x):
     # Not actually object dtype, just need to specify something
     tmp = elemwise(np.frexp, x, dtype=object)
@@ -159,9 +239,7 @@ def frexp(x):
     return L, R
 
 
-frexp.__doc__ = skip_doctest(np.frexp.__doc__)
-
-
+@copy_docstring(source=np.modf)
 def modf(x):
     # Not actually object dtype, just need to specify something
     tmp = elemwise(np.modf, x, dtype=object)
@@ -180,6 +258,3 @@ def modf(x):
     L = Array(sharedict.merge(tmp.dask, (left, ldsk)), left, chunks=tmp.chunks, dtype=ldt)
     R = Array(sharedict.merge(tmp.dask, (right, rdsk)), right, chunks=tmp.chunks, dtype=rdt)
     return L, R
-
-
-modf.__doc__ = skip_doctest(np.modf.__doc__)
diff --git a/dask/bytes/core.py b/dask/bytes/core.py
index 975e7d44f..45d731f09 100644
--- a/dask/bytes/core.py
+++ b/dask/bytes/core.py
@@ -12,7 +12,7 @@ from .utils import (SeekableFile, read_block, infer_compression,
 from ..compatibility import PY2, unicode
 from ..base import tokenize, normalize_token
 from ..delayed import delayed
-from ..utils import import_required, ensure_bytes, ensure_unicode
+from ..utils import import_required, ensure_bytes, ensure_unicode, is_integer
 
 
 def write_block_to_file(data, lazy_file):
@@ -138,6 +138,11 @@ def read_bytes(urlpath, delimiter=None, not_zero=False, blocksize=2**27,
     if len(paths) == 0:
         raise IOError("%s resolved to no files" % urlpath)
 
+    if blocksize is not None:
+        if not is_integer(blocksize):
+            raise TypeError("blocksize must be an integer")
+        blocksize = int(blocksize)
+
     blocks, lengths, machines = fs.get_block_locations(paths)
     if blocks:
         offsets = blocks
diff --git a/dask/utils.py b/dask/utils.py
index b519a7eca..637f7bd82 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -11,12 +11,13 @@ from errno import ENOENT
 from collections import Iterator
 from contextlib import contextmanager
 from importlib import import_module
+from numbers import Integral
 from threading import Lock
 import multiprocessing as mp
 import uuid
 from weakref import WeakValueDictionary
 
-from .compatibility import long, getargspec, PY3, unicode, urlsplit
+from .compatibility import getargspec, PY3, unicode, urlsplit
 from .core import get_deps
 from .context import _globals
 from .optimize import key_split    # noqa: F401
@@ -285,15 +286,7 @@ def is_integer(i):
     >>> is_integer('abc')
     False
     """
-    import numpy as np
-    if isinstance(i, (int, long)):
-        return True
-    if isinstance(i, float):
-        return (i).is_integer()
-    if issubclass(type(i), np.integer):
-        return i
-    else:
-        return False
+    return isinstance(i, Integral) or (isinstance(i, float) and i.is_integer())
 
 
 ONE_ARITY_BUILTINS = set([abs, all, any, bool, bytearray, bytes, callable, chr,


Test patch:
diff --git a/dask/array/tests/test_ufunc.py b/dask/array/tests/test_ufunc.py
index ceccd6c32..3f777e219 100644
--- a/dask/array/tests/test_ufunc.py
+++ b/dask/array/tests/test_ufunc.py
@@ -18,16 +18,39 @@ def test_ufunc_meta():
     assert da.frexp.__doc__.replace('    # doctest: +SKIP', '') == np.frexp.__doc__
 
 
-@pytest.mark.parametrize('ufunc',
-                         ['conj', 'exp', 'exp2', 'log', 'log2', 'log10',
-                          'log1p', 'expm1', 'sqrt', 'cbrt', 'square', 'sin',
-                          'cos', 'tan', 'arctan', 'sinh', 'cosh', 'tanh',
-                          'arcsinh', 'arccosh', 'deg2rad', 'rad2deg',
-                          'isfinite', 'isinf', 'isnan', 'signbit', 'i0',
-                          'reciprocal', 'degrees', 'radians', 'rint', 'fabs',
-                          'sign', 'absolute', 'floor', 'ceil', 'trunc',
-                          'logical_not', 'spacing', 'sinc', 'nan_to_num'])
-def test_ufunc(ufunc):
+def test_ufunc():
+    for attr in ['nin', 'nargs', 'nout', 'ntypes', 'identity',
+                 'signature', 'types']:
+        assert getattr(da.log, attr) == getattr(np.log, attr)
+
+    with pytest.raises(AttributeError):
+        da.log.not_an_attribute
+
+    assert repr(da.log) == repr(np.log)
+    assert 'nin' in dir(da.log)
+    assert 'outer' in dir(da.log)
+
+
+binary_ufuncs = ['add', 'arctan2', 'copysign', 'divide', 'equal',
+                 'floor_divide', 'fmax', 'fmin', 'fmod', 'greater',
+                 'greater_equal', 'hypot', 'ldexp', 'less', 'less_equal',
+                 'logaddexp', 'logaddexp2', 'logical_and', 'logical_or',
+                 'logical_xor', 'maximum', 'minimum', 'mod', 'multiply',
+                 'nextafter', 'not_equal', 'power', 'remainder', 'subtract',
+                 'true_divide']
+
+unary_ufuncs = ['absolute', 'arccos', 'arccosh', 'arcsin', 'arcsinh', 'arctan',
+                'arctanh', 'cbrt', 'ceil', 'conj', 'cos', 'cosh', 'deg2rad',
+                'degrees', 'exp', 'exp2', 'expm1', 'fabs', 'fix', 'floor',
+                'i0', 'isfinite', 'isinf', 'isnan', 'log', 'log10', 'log1p',
+                'log2', 'logical_not', 'nan_to_num', 'negative', 'rad2deg',
+                'radians', 'reciprocal', 'rint', 'sign', 'signbit', 'sin',
+                'sinc', 'sinh', 'spacing', 'sqrt', 'square', 'tan', 'tanh',
+                'trunc']
+
+
+@pytest.mark.parametrize('ufunc', unary_ufuncs)
+def test_unary_ufunc(ufunc):
 
     dafunc = getattr(da, ufunc)
     npfunc = getattr(np, ufunc)
@@ -37,27 +60,19 @@ def test_ufunc(ufunc):
 
     # applying Dask ufunc doesn't trigger computation
     assert isinstance(dafunc(darr), da.Array)
-    assert_eq(dafunc(darr), npfunc(arr))
+    assert_eq(dafunc(darr), npfunc(arr), equal_nan=True)
 
     # applying NumPy ufunc triggers computation
     assert isinstance(npfunc(darr), np.ndarray)
-    assert_eq(npfunc(darr), npfunc(arr))
+    assert_eq(npfunc(darr), npfunc(arr), equal_nan=True)
 
     # applying Dask ufunc to normal ndarray triggers computation
     assert isinstance(dafunc(arr), np.ndarray)
-    assert_eq(dafunc(arr), npfunc(arr))
+    assert_eq(dafunc(arr), npfunc(arr), equal_nan=True)
 
 
-@pytest.mark.parametrize('ufunc', ['add', 'subtract', 'multiply', 'divide',
-                                   'true_divide', 'floor_divide', 'power',
-                                   'remainder', 'mod', 'fmax', 'fmin',
-                                   'logaddexp', 'logaddexp2', 'arctan2',
-                                   'hypot', 'copysign', 'nextafter', 'ldexp',
-                                   'fmod', 'logical_and', 'logical_or',
-                                   'logical_xor', 'maximum', 'minimum',
-                                   'greater', 'greater_equal', 'less',
-                                   'less_equal', 'not_equal', 'equal'])
-def test_ufunc_2args(ufunc):
+@pytest.mark.parametrize('ufunc', binary_ufuncs)
+def test_binary_ufunc(ufunc):
 
     dafunc = getattr(da, ufunc)
     npfunc = getattr(np, ufunc)
@@ -94,6 +109,45 @@ def test_ufunc_2args(ufunc):
     assert_eq(dafunc(10, arr1), npfunc(10, arr1))
 
 
+def test_ufunc_outer():
+    arr1 = np.random.randint(1, 100, size=20)
+    darr1 = da.from_array(arr1, 3)
+
+    arr2 = np.random.randint(1, 100, size=(10, 3))
+    darr2 = da.from_array(arr2, 3)
+
+    # Check output types
+    assert isinstance(da.add.outer(darr1, darr2), da.Array)
+    assert isinstance(da.add.outer(arr1, darr2), da.Array)
+    assert isinstance(da.add.outer(darr1, arr2), da.Array)
+    assert isinstance(da.add.outer(arr1, arr2), np.ndarray)
+
+    # Check mix of dimensions, dtypes, and numpy/dask/object
+    cases = [((darr1, darr2), (arr1, arr2)),
+             ((darr2, darr1), (arr2, arr1)),
+             ((darr2, darr1.astype('f8')), (arr2, arr1.astype('f8'))),
+             ((darr1, arr2), (arr1, arr2)),
+             ((darr1, 1), (arr1, 1)),
+             ((1, darr2), (1, arr2)),
+             ((1.5, darr2), (1.5, arr2)),
+             (([1, 2, 3], darr2), ([1, 2, 3], arr2)),
+             ((darr1.sum(), darr2), (arr1.sum(), arr2)),
+             ((np.array(1), darr2), (np.array(1), arr2))]
+
+    for (dA, dB), (A, B) in cases:
+        assert_eq(da.add.outer(dA, dB), np.add.outer(A, B))
+
+    # Check dtype kwarg works
+    assert_eq(da.add.outer(darr1, darr2, dtype='f8'),
+              np.add.outer(arr1, arr2, dtype='f8'))
+
+    with pytest.raises(ValueError):
+        da.add.outer(darr1, darr2, out=arr1)
+
+    with pytest.raises(ValueError):
+        da.sin.outer(darr1, darr2)
+
+
 @pytest.mark.parametrize('ufunc', ['isreal', 'iscomplex', 'real', 'imag'])
 def test_complex(ufunc):
 
@@ -170,3 +224,15 @@ def test_clip():
     assert_eq(x.clip(max=5), d.clip(max=5))
     assert_eq(x.clip(max=1, min=5), d.clip(max=1, min=5))
     assert_eq(x.clip(min=1, max=5), d.clip(min=1, max=5))
+
+
+def test_angle():
+    real = np.random.randint(1, 100, size=(20, 20))
+    imag = np.random.randint(1, 100, size=(20, 20)) * 1j
+    comp = real + imag
+    dacomp = da.from_array(comp, 3)
+
+    assert_eq(da.angle(dacomp), np.angle(comp))
+    assert_eq(da.angle(dacomp, deg=True), np.angle(comp, deg=True))
+    assert isinstance(da.angle(comp), np.ndarray)
+    assert_eq(da.angle(comp), np.angle(comp))
diff --git a/dask/bytes/tests/test_local.py b/dask/bytes/tests/test_local.py
index 5ab3b2e22..79ccab5a5 100644
--- a/dask/bytes/tests/test_local.py
+++ b/dask/bytes/tests/test_local.py
@@ -62,6 +62,18 @@ def test_read_bytes_blocksize_none():
         assert sum(map(len, values)) == len(files)
 
 
+def test_read_bytes_blocksize_float():
+    with filetexts(files, mode='b'):
+        sample, vals = read_bytes('.test.account*', blocksize=5.0)
+        results = compute(*concat(vals))
+        ourlines = b"".join(results).split(b'\n')
+        testlines = b"".join(files.values()).split(b'\n')
+        assert set(ourlines) == set(testlines)
+
+        with pytest.raises(TypeError):
+            read_bytes('.test.account*', blocksize=5.5)
+
+
 def test_with_urls():
     with filetexts(files, mode='b'):
         # OS-independent file:// URI with glob *