Dask support (#4)

* feat: Add support for dask arrays

Serialization wil be done by block (only split the first dim by 1st dim chunk size)

* fix: Yield each elements od dods_encode

use to yield the generator itself...

* Merge branch 'origin/dask-support' into local branch 'dask-suppoert'

* fix: allow for np.int64 data

* fix: added missing requirements

* test: added Python 3.9. to both tox.ini and travis config

* doc: Adapted README to reflect dask.Array support

Co-authored-by: ber <[email protected]>
Co-authored-by: Philipp Falke <[email protected]>
Co-authored-by: Philipp Falke <[email protected]>

Author: 3 people

.travis.yml

@@ -3,6 +3,7 @@ python:
 - '3.6'
 - '3.7'
 - '3.8'
+- '3.9'
 install:
 - python setup.py install
 - pip install codecov

README.md

@@ -12,4 +12,4 @@ A pure Python implementation of the OPeNDAP server protocol.
 This module allows you to serve arbitrary data structures through the web
 framework of your choice as OPeNDAP data objects. It implements just the bare
 minimum of the DAP 2.0 protocol: DDS, DAS, and DODS responses and slicing. Array
-data needs to be supplied as `numpy.ndarray`.
+data needs to be supplied as `numpy.ndarray` or as `dask.array.Array`.

opendap_protocol/protocol.py

@@ -25,7 +25,6 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
 """
 A pure Python implementation of the OPeNDAP server protocol.
 
@@ -43,6 +42,7 @@
 
 import re
 import numpy as np
+import dask.array as da
 
 INDENT = '    '
 SLICE_CONSTRAINT_RE = r'\[([\d,\W]+)\]$'
@@ -55,7 +55,6 @@ class DAPError(Exception):
 class DAPObject(object):
     """A generic DAP object class.
     """
-
     def __init__(self, name='', parent=None, *args, **kwargs):
         try:
             self.name = '_'.join(name.split(' '))
@@ -132,12 +131,12 @@ def data_path(self):
                 return '.'.join([self.parent.data_path, self.name])
 
     def ddshead(self):
-        return '{indent}{obj} {{\n'.format(
-            indent=self.indent, obj=self.__class__.__name__)
+        return '{indent}{obj} {{\n'.format(indent=self.indent,
+                                           obj=self.__class__.__name__)
 
     def ddstail(self):
-        return '{indent}}} {name};\n'.format(
-            indent=self.indent, name=self.name)
+        return '{indent}}} {name};\n'.format(indent=self.indent,
+                                             name=self.name)
 
     def dashead(self):
         name = self.name
@@ -211,12 +210,13 @@ def das(self, constraint=''):
 
     def dds(self, constraint=''):
         if meets_constraint(constraint, self.data_path):
-            yield '{indent}{dtype} {name};\n'.format(
-                indent=self.indent, dtype=self.__str__(), name=self.name)
+            yield '{indent}{dtype} {name};\n'.format(indent=self.indent,
+                                                     dtype=self.__str__(),
+                                                     name=self.name)
 
     def dods_data(self, constraint=''):
         if meets_constraint(constraint, self.data_path):
-            yield dods_encode(self._val, self)
+            yield from dods_encode(self._val, self)
 
 
 class Byte(DAPAtom):
@@ -226,32 +226,32 @@ class Byte(DAPAtom):
 
 class Int16(DAPAtom):
     dtype = np.int16
-    str = '<i4'
+    str = '>i4'
 
 
 class UInt16(DAPAtom):
     dtype = np.uint16
-    str = '<u4'
+    str = '>u4'
 
 
 class Int32(DAPAtom):
     dtype = np.int32
-    str = '<i4'
+    str = '>i4'
 
 
 class UInt32(DAPAtom):
     dtype = np.uint32
-    str = '<u4'
+    str = '>u4'
 
 
 class Float32(DAPAtom):
     dtype = np.float32
-    str = '<f4'
+    str = '>f4'
 
 
 class Float64(DAPAtom):
     dtype = np.float64
-    str = '<f8'
+    str = '>f8'
 
 
 class String(DAPAtom):
@@ -260,7 +260,7 @@ class String(DAPAtom):
 
     def dods_data(self, constraint=''):
         if meets_constraint(constraint, self.data_path):
-            yield dods_encode(self._val.encode('ascii'), self)
+            yield from dods_encode(self._val.encode('ascii'), self)
 
 
 class URL(String):
@@ -277,7 +277,6 @@ class Structure(DAPObject):
 class Dataset(Structure):
     """Class representing a DAP dataset.
     """
-
     def dods_data(self, constraint=''):
 
         yield b'Data:\r\n'
@@ -342,7 +341,6 @@ def dods_data(self, constraint=''):
 class SequenceInstance(DAPObject):
     """Class representing a data item that will be added to a sequence.
     """
-
     @property
     def data_path(self):
         return self.parent.data_path
@@ -372,7 +370,6 @@ class DAPDataObject(DAPObject):
     """A generic class for typed non-atomic objects holding actual data (i.e.
     Array and Grid).
     """
-
     def _parse_args(self, args, kwargs):
 
         self.data = kwargs.get('data', None)
@@ -394,11 +391,11 @@ def dods_data(self, constraint=''):
 
         if meets_constraint(constraint, self.data_path):
             slices = parse_slice_constraint(constraint)
-            yield dods_encode(self.data[slices], self.dtype)
+            yield from dods_encode(self.data[slices], self.dtype)
             if self.dimensions is not None:
                 for i, dim in enumerate(self.dimensions):
                     sl = slices[i] if i < len(slices) else ...
-                    yield dods_encode(dim.data[sl], dim.dtype)
+                    yield from dods_encode(dim.data[sl], dim.dtype)
 
 
 class Grid(DAPDataObject):
@@ -482,9 +479,25 @@ def dods_encode(data, dtype):
     if not is_scalar:
         packed_length = length.astype('<i4').byteswap().tobytes() * 2
 
-    packed_data = data.astype(dtype.str).byteswap().tobytes()
 
-    return packed_length + packed_data
+    yield packed_length
+
+
+    if isinstance(data, da.Array):
+        for x in range(0, data.shape[0], data.chunks[0][0]):
+            yield np.array(data[x:x + data.chunks[0][0],
+                                ...]).astype(dtype.str).tobytes()
+    else:
+        yield data.astype(dtype.str).tobytes()
+        #yield data.astype(dtype.str).byteswap().tobytes()
+
+    ######
+    #import pudb; pudb.set_trace()
+    #if isinstance(data, dask.array.Array):
+    #    return packed_length + data.map_blocks(np.ndarray.tobytes, dtype=dtype.str)
+    #else:
+    #    packed_data = data.astype(dtype.str).tobytes()
+    #    return packed_length + packed_data
 
 
 def parse_slice_constraint(constraint):

setup.py

@@ -33,6 +33,8 @@
 
 requirements = [
     'numpy',
+    'dask',
+    'toolz',
 ]
 setup_requirements = [
     'setuptools_scm',

tests/test_all.py

@@ -26,10 +26,13 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+from opendap_protocol.protocol import dods_encode
 import opendap_protocol as dap
 
 import xdrlib
 import numpy as np
+import dask.array as da
+from functools import reduce
 
 XDRPACKER = xdrlib.Packer()
 
@@ -40,13 +43,37 @@ def test_dods_encode():
 
     xdrpacked = pack_xdr_float(testdata)
 
-    assert xdrpacked == dap.dods_encode(testdata, dap.Float32)
+    assert xdrpacked == b''.join(dap.dods_encode(testdata, dap.Float32))
 
-    assert b'\x00\x00\x00\x00' == dap.dods_encode(0, dap.Float32)
+    assert b'\x00\x00\x00\x00' == b''.join(dap.dods_encode(0, dap.Float32))
 
     arrdata = np.asarray([1, 2, 3])
-    assert dap.dods_encode(arrdata,
-                           dap.Float64) == pack_xdr_double_array(arrdata)
+    assert b''.join(dap.dods_encode(
+        arrdata, dap.Float64)) == pack_xdr_double_array(arrdata)
+
+    # test dask vs numpy
+    x_dim = 28
+    y_dim = 30
+    time_dim = 8
+    vertical_dim = 1
+    real_dim = 21
+    ref_time_dim = 3
+
+    np_data = np.arange(
+        0, x_dim * y_dim * time_dim * vertical_dim * real_dim *
+        ref_time_dim).reshape(
+            (x_dim, y_dim, time_dim, vertical_dim, real_dim, ref_time_dim))
+
+    data_vals = da.from_array(np_data,
+                              chunks=(14, y_dim, 1, vertical_dim, 1, 1))
+
+    x = dap.dods_encode(data_vals, dap.Int32)
+    y = dap.dods_encode(np_data, dap.Int32)
+    assert b''.join(x) == b''.join(y)
+
+    int_arrdata = np.arange(0, 20, 2, dtype='<i4')
+    assert b''.join(dods_encode(int_arrdata,
+                                dap.Int32)) == pack_xdr_int_array(int_arrdata)
 
 
 def test_parse_slice():
@@ -58,7 +85,8 @@ def test_parse_slice():
 
 def test_parse_slice_constraint():
 
-    assert dap.parse_slice_constraint('[0][:][:][4:7]') == (0, Ellipsis, Ellipsis,
+    assert dap.parse_slice_constraint('[0][:][:][4:7]') == (0, Ellipsis,
+                                                            Ellipsis,
                                                             slice(4, 8))
     assert dap.parse_slice_constraint('[0][:][:]') == (0, Ellipsis, Ellipsis)
     assert dap.parse_slice_constraint('[0][:]') == (0, Ellipsis)
@@ -111,8 +139,9 @@ def test_Attribute():
     dataset = dap.Dataset(name='test')
 
     attr1 = dap.Attribute(name='Attribute 1', value=3, dtype=dap.Float32)
-    attr2 = dap.Attribute(
-        name='Attribute 2', value='a string', dtype=dap.String)
+    attr2 = dap.Attribute(name='Attribute 2',
+                          value='a string',
+                          dtype=dap.String)
 
     dataset.append(attr1, attr2)
 
@@ -160,11 +189,10 @@ def test_complete_dap_response():
     x = dap.Array(name='x', data=np.array([0, 1]), dtype=dap.Int16)
     y = dap.Array(name='y', data=np.array([10, 11]), dtype=dap.Int16)
 
-    z = dap.Grid(
-        name='z',
-        data=np.array([[0, 0], [0, 0]]),
-        dtype=dap.Int32,
-        dimensions=[x, y])
+    z = dap.Grid(name='z',
+                 data=np.array([[0, 0], [0, 0]]),
+                 dtype=dap.Int32,
+                 dimensions=[x, y])
 
     z_attr = [
         dap.Attribute(name='units', value='second', dtype=dap.String),
@@ -197,3 +225,11 @@ def pack_xdr_double_array(data):
     XDRPACKER.pack_int(np.asarray(len(data)))
     XDRPACKER.pack_farray(len(data), data.astype('<f8'), XDRPACKER.pack_double)
     return XDRPACKER.get_buffer()
+
+
+def pack_xdr_int_array(data):
+    XDRPACKER.reset()
+    XDRPACKER.pack_int(np.asarray(len(data)))
+    XDRPACKER.pack_int(np.asarray(len(data)))
+    XDRPACKER.pack_farray(len(data), data.astype('<i4'), XDRPACKER.pack_int)
+    return XDRPACKER.get_buffer()

tox.ini

@@ -4,7 +4,7 @@
 # and then run "tox" from this directory.
 
 [tox]
-envlist = clean, py36, py37, py38, coverage
+envlist = clean, py36, py37, py38, py39, coverage
 
 [testenv]