Adding a to_spatial method on ParquetSource

.travis.yml

@@ -5,6 +5,7 @@ env:
   matrix:
   - CONDA_PY=36
   global:
+    CHANNELS="-c intake -c defaults -c conda-forge"
     secure: ujj1B51Q6iG0f65hMj1bLUNgxHN0rh3dBKkQb3dELJ+OzU8r6YBXjrZp6PS0OM1PGbR5tcLwfmZG8u9QR1NTDAfYlVKXcd86SWqZl3XTQT7IBn7Z0aqV4ILJfpxacSm7Uu7lu9IKAySrDhhi+k4HbUkYk+xEL72e0F2QS07QldmPU3poCxJfpGiavGumiK6JzQ/Pmln+y6SYCtbf/6E4Ub/9jmBJ6XwyXAXS1zaUCbCnaMnLox3WIDaOInSmAa+yUh6E7uWAMOYF3w89ZGa5Q69TXLVZitUDRut1Faouu8RZbAsCdRHtR1CtqciLZiUf0kaWibDXpKoCQcnQIliDGbZIEBa191xOB8joTtb8F2Hz4QhzQ8tBZDa9hxOA/XlGFmBvNh27eMHO2PfJu08s3S/FZGaMjHuGsECvOmVor6lcHPqSE7gmRGUxiH3VQ++agtGw4I8qD83fsKRvnCR+YopViEgtjMAaXHWSNa4AMB4ITFDGx7omnGSpcuVAQ0AmGM8qk8GqhDYzAzMmOmzPYGxFCcvbciFVcwwnsbQ4iMI9XFhCP2Sd1XVzKEGzOMhKDb0JI+3jAb32mlMpcyGhmi+k3smLzgtZTf2vzH0hq5saRrc4z8IzSsxQTPeNkT0s9O7lo8VujI97dBXOC5RdSRohXR8As6AVkSc9bpJYR4I=
 install:
 - |
@@ -20,7 +21,10 @@ install:
 script:
 - |
   flake8 .
-  conda build -c intake -c defaults -c conda-forge ./conda
+  conda build $CHANNELS ./conda --no-test
+- conda install -y $CHANNELS 'datashader==0.7.0' pytest pytest-cov coverage
+- conda install -y $CHANNELS --use-local intake_parquet
+- py.test --verbose --cov=intake_parquet tests
 - |
   if [ -n "$TRAVIS_TAG" ]; then
       # If tagged git version, upload package to main channel

intake_parquet/source.py

@@ -106,5 +106,29 @@ def _to_dask(self):
         self._load_metadata()
         return self._df
 
+    def to_spatial(self):
+        """Read a data source as a SpatialPointsFrame.
+
+        This requires that the data be structured in a particular, spatially
+        optimized, way. To convert a regular dataframe to a SpatialPointsFrame
+        use the functionality in datshader.
+
+        >>> import datashader.spatial.points as dsp
+        >>> dsp.to_parquet(df, 'sorted.parq', 'x', 'y', shuffle='disk', npartitions=32)
+
+        More info at http://datashader.org/user_guide/2_Points.html
+
+        NOTE: This method only works for local or cached data.
+        """
+        try:
+            from datashader.spatial import read_parquet
+        except ImportError:
+            raise ImportError('SpatialPointsFrame not found in this '
+                'version of datashader. Get latest using '
+                '`conda install -c pyviz datashader`.')
+
+        urlpath = self._get_cache(self._urlpath)[0]
+        return read_parquet(urlpath)
+
     def _close(self):
         self._df = None

tests/test_spatial.py

@@ -0,0 +1,76 @@
+"""
+Minimal spatial test copied from datashader
+"""
+import os
+import pytest
+import numpy as np
+import pandas as pd
+
+from intake_parquet.source import ParquetSource
+import intake
+
+pytest.importorskip('datashader')
+
+intake.registry['parquet'] = ParquetSource  # because pytest defers import
+
+@pytest.fixture()
+def df():
+    N = 1000
+    np.random.seed(25)
+
+    df = pd.DataFrame({
+        'x': np.random.rand(N),
+        'y': np.random.rand(N) * 2,
+        'a': np.random.randn(N)
+    })
+
+    # Make sure we have x/y values of 0 and 1 represented so that
+    # autocomputed ranges are predictable
+    df.x.iloc[0] = 0.0
+    df.x.iloc[-1] = 1.0
+    df.y.iloc[0] = 0.0
+    df.y.iloc[-1] = 2.0
+    return df
+
+@pytest.fixture(params=[False, True])
+def s_points_frame(request, tmp_path, df):
+    import datashader.spatial.points as dsp
+
+    # Work around https://bugs.python.org/issue33617
+    tmp_path = str(tmp_path)
+    p = 5
+    path = os.path.join(tmp_path, 'spatial_points.parquet')
+
+    dsp.to_parquet(
+        df, path, 'x', 'y', p=p, npartitions=10)
+
+    spf = ParquetSource(path).to_spatial()
+
+    if request.param:
+        spf = spf.persist()
+
+    return spf
+
+
+def test_spatial_points_frame_properties(s_points_frame):
+    assert s_points_frame.spatial.x == 'x'
+    assert s_points_frame.spatial.y == 'y'
+    assert s_points_frame.spatial.p == 5
+    assert s_points_frame.npartitions == 10
+    assert s_points_frame.spatial.x_range == (0, 1)
+    assert s_points_frame.spatial.y_range == (0, 2)
+    assert s_points_frame.spatial.nrows == 1000
+
+    # x_bin_edges
+    np.testing.assert_array_equal(
+        s_points_frame.spatial.x_bin_edges,
+        np.linspace(0.0, 1.0, 2 ** 5 + 1))
+
+    # y_bin_edges
+    np.testing.assert_array_equal(
+        s_points_frame.spatial.y_bin_edges,
+        np.linspace(0.0, 2.0, 2 ** 5 + 1))
+
+    # distance_divisions
+    distance_divisions = s_points_frame.spatial.distance_divisions
+    assert len(distance_divisions) == 10 + 1