File: C:/Users/fred/anaconda3/Lib/site-packages/holoviews/tests/core/test_datasetproperty.py
from unittest import SkipTest
import numpy as np
import pandas as pd
try:
import dask.dataframe as dd
except ImportError:
dd = None
from holoviews import Curve, Dataset, Dimension, Distribution, Scatter
from holoviews.core import Apply, Redim
from holoviews.element.comparison import ComparisonTestCase
from holoviews.operation import function, histogram
try:
from holoviews.operation.datashader import datashade, dynspread, rasterize
except ImportError:
dynspread = datashade = rasterize = None
class DatasetPropertyTestCase(ComparisonTestCase):
def setUp(self):
self.df = pd.DataFrame({
'a': [1, 1, 3, 3, 2, 2, 0, 0],
'b': [10, 20, 30, 40, 10, 20, 30, 40],
'c': ['A', 'A', 'B', 'B', 'C', 'C', 'D', 'D'],
'd': [-1, -2, -3, -4, -5, -6, -7, -8]
})
self.ds = Dataset(
self.df,
kdims=[
Dimension('a', label="The a Column"),
Dimension('b', label="The b Column"),
Dimension('c', label="The c Column"),
Dimension('d', label="The d Column"),
]
)
self.ds2 = Dataset(
self.df.iloc[2:],
kdims=[
Dimension('a', label="The a Column"),
Dimension('b', label="The b Column"),
Dimension('c', label="The c Column"),
Dimension('d', label="The d Column"),
]
)
class ConstructorTestCase(DatasetPropertyTestCase):
def test_constructors_dataset(self):
ds = Dataset(self.df)
self.assertIs(ds, ds.dataset)
# Check pipeline
ops = ds.pipeline.operations
self.assertEqual(len(ops), 1)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ds, ds.pipeline(ds.dataset))
def test_constructor_curve(self):
element = Curve(self.df)
expected = Dataset(
self.df,
kdims=self.df.columns[0],
vdims=self.df.columns[1:].tolist(),
)
self.assertEqual(element.dataset, expected)
# Check pipeline
pipeline = element.pipeline
self.assertEqual(len(pipeline.operations), 1)
self.assertIs(pipeline.operations[0].output_type, Curve)
self.assertEqual(element, element.pipeline(element.dataset))
class ToTestCase(DatasetPropertyTestCase):
def test_to_element(self):
curve = self.ds.to(Curve, 'a', 'b', groupby=[])
curve2 = self.ds2.to(Curve, 'a', 'b', groupby=[])
self.assertNotEqual(curve, curve2)
self.assertEqual(curve.dataset, self.ds)
scatter = curve.to(Scatter)
self.assertEqual(scatter.dataset, self.ds)
# Check pipeline
ops = curve.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
self.assertEqual(
curve.pipeline(self.ds2), curve2
)
def test_to_holomap(self):
curve_hmap = self.ds.to(Curve, 'a', 'b', groupby=['c'])
# Check HoloMap element datasets
for v in self.df.c.drop_duplicates():
curve = curve_hmap.data[(v,)]
# check dataset
self.assertEqual(
curve.dataset, self.ds
)
# execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
def test_to_holomap_dask(self):
if dd is None:
raise SkipTest("Dask required to test .to with dask dataframe.")
ddf = dd.from_pandas(self.df, npartitions=2)
dds = Dataset(
ddf,
kdims=[
Dimension('a', label="The a Column"),
Dimension('b', label="The b Column"),
Dimension('c', label="The c Column"),
Dimension('d', label="The d Column"),
]
)
curve_hmap = dds.to(Curve, 'a', 'b', groupby=['c'])
# Check HoloMap element datasets
for v in self.df.c.drop_duplicates():
curve = curve_hmap.data[(v,)]
self.assertEqual(
curve.dataset, self.ds
)
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
class CloneTestCase(DatasetPropertyTestCase):
def test_clone(self):
# Dataset
self.assertEqual(self.ds.clone().dataset, self.ds)
# Curve
curve = self.ds.to.curve('a', 'b', groupby=[])
curve_clone = curve.clone()
self.assertEqual(
curve_clone.dataset,
self.ds
)
# Check pipeline carried over
self.assertEqual(
curve.pipeline.operations, curve_clone.pipeline.operations[:2]
)
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
def test_clone_new_data(self):
# Replacing data during clone resets .dataset
ds_clone = self.ds.clone(data=self.ds2.data)
self.assertEqual(ds_clone.dataset, self.ds2)
self.assertEqual(len(ds_clone.pipeline.operations), 1)
def test_clone_dataset_kwarg_none(self):
# Setting dataset=None prevents propagation of dataset to cloned object
ds_clone = self.ds.clone(dataset=None)
self.assertIs(ds_clone, ds_clone.dataset)
class ReindexTestCase(DatasetPropertyTestCase):
def test_reindex_dataset(self):
ds_ab = self.ds.reindex(kdims=['a'], vdims=['b'])
ds2_ab = self.ds2.reindex(kdims=['a'], vdims=['b'])
self.assertNotEqual(ds_ab, ds2_ab)
self.assertEqual(ds_ab.dataset, self.ds)
# Check pipeline
ops = ds_ab.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'reindex')
self.assertEqual(ops[1].args, [])
self.assertEqual(ops[1].kwargs, dict(kdims=['a'], vdims=['b']))
# Execute pipeline
self.assertEqual(ds_ab.pipeline(ds_ab.dataset), ds_ab)
self.assertEqual(
ds_ab.pipeline(self.ds2), ds2_ab
)
def test_double_reindex_dataset(self):
ds_ab = (self.ds
.reindex(kdims=['a'], vdims=['b', 'c'])
.reindex(kdims=['a'], vdims=['b']))
ds2_ab = (self.ds2
.reindex(kdims=['a'], vdims=['b', 'c'])
.reindex(kdims=['a'], vdims=['b']))
self.assertNotEqual(ds_ab, ds2_ab)
self.assertEqual(ds_ab.dataset, self.ds)
# Check pipeline
ops = ds_ab.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'reindex')
self.assertEqual(ops[1].args, [])
self.assertEqual(ops[1].kwargs, dict(kdims=['a'], vdims=['b', 'c']))
self.assertEqual(ops[2].method_name, 'reindex')
self.assertEqual(ops[2].args, [])
self.assertEqual(ops[2].kwargs, dict(kdims=['a'], vdims=['b']))
# Execute pipeline
self.assertEqual(ds_ab.pipeline(ds_ab.dataset), ds_ab)
self.assertEqual(
ds_ab.pipeline(self.ds2), ds2_ab
)
def test_reindex_curve(self):
curve_ba = self.ds.to(
Curve, 'a', 'b', groupby=[]
).reindex(kdims='b', vdims='a')
curve2_ba = self.ds2.to(
Curve, 'a', 'b', groupby=[]
).reindex(kdims='b', vdims='a')
self.assertNotEqual(curve_ba, curve2_ba)
self.assertEqual(curve_ba.dataset, self.ds)
# Check pipeline
ops = curve_ba.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, 'reindex')
self.assertEqual(ops[2].args, [])
self.assertEqual(ops[2].kwargs, dict(kdims='b', vdims='a'))
# Execute pipeline
self.assertEqual(curve_ba.pipeline(curve_ba.dataset), curve_ba)
self.assertEqual(
curve_ba.pipeline(self.ds2), curve2_ba
)
def test_double_reindex_curve(self):
curve_ba = self.ds.to(
Curve, 'a', ['b', 'c'], groupby=[]
).reindex(kdims='a', vdims='b').reindex(kdims='b', vdims='a')
curve2_ba = self.ds2.to(
Curve, 'a', ['b', 'c'], groupby=[]
).reindex(kdims='a', vdims='b').reindex(kdims='b', vdims='a')
self.assertNotEqual(curve_ba, curve2_ba)
self.assertEqual(curve_ba.dataset, self.ds)
# Check pipeline
ops = curve_ba.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, 'reindex')
self.assertEqual(ops[2].args, [])
self.assertEqual(ops[2].kwargs, dict(kdims='a', vdims='b'))
self.assertEqual(ops[3].method_name, 'reindex')
self.assertEqual(ops[3].args, [])
self.assertEqual(ops[3].kwargs, dict(kdims='b', vdims='a'))
# Execute pipeline
self.assertEqual(curve_ba.pipeline(curve_ba.dataset), curve_ba)
self.assertEqual(
curve_ba.pipeline(self.ds2), curve2_ba
)
class IlocTestCase(DatasetPropertyTestCase):
def test_iloc_dataset(self):
ds_iloc = self.ds.iloc[[0, 2]]
ds2_iloc = self.ds2.iloc[[0, 2]]
self.assertNotEqual(ds_iloc, ds2_iloc)
# Dataset
self.assertEqual(
ds_iloc.dataset,
self.ds
)
# Check pipeline
ops = ds_iloc.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, '_perform_getitem')
self.assertEqual(ops[1].args, [[0, 2]])
self.assertEqual(ops[1].kwargs, {})
# Execute pipeline
self.assertEqual(ds_iloc.pipeline(ds_iloc.dataset), ds_iloc)
self.assertEqual(
ds_iloc.pipeline(self.ds2), ds2_iloc
)
def test_iloc_curve(self):
# Curve
curve_iloc = self.ds.to.curve('a', 'b', groupby=[]).iloc[[0, 2]]
curve2_iloc = self.ds2.to.curve('a', 'b', groupby=[]).iloc[[0, 2]]
self.assertNotEqual(curve_iloc, curve2_iloc)
self.assertEqual(
curve_iloc.dataset,
self.ds
)
# Check pipeline
ops = curve_iloc.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, '_perform_getitem')
self.assertEqual(ops[2].args, [[0, 2]])
self.assertEqual(ops[2].kwargs, {})
# Execute pipeline
self.assertEqual(curve_iloc.pipeline(curve_iloc.dataset), curve_iloc)
self.assertEqual(
curve_iloc.pipeline(self.ds2), curve2_iloc
)
class NdlocTestCase(DatasetPropertyTestCase):
def setUp(self):
super().setUp()
self.ds_grid = Dataset(
(np.arange(4),
np.arange(3),
np.array([[1, 2, 3, 4],
[5, 6, 7, 8],
[9, 10, 11, 12]])),
kdims=['x', 'y'],
vdims='z'
)
self.ds2_grid = Dataset(
(np.arange(3),
np.arange(3),
np.array([[1, 2, 4],
[5, 6, 8],
[9, 10, 12]])),
kdims=['x', 'y'],
vdims='z'
)
def test_ndloc_dataset(self):
ds_grid_ndloc = self.ds_grid.ndloc[0:2, 1:3]
ds2_grid_ndloc = self.ds2_grid.ndloc[0:2, 1:3]
self.assertNotEqual(ds_grid_ndloc, ds2_grid_ndloc)
# Dataset
self.assertEqual(
ds_grid_ndloc.dataset,
self.ds_grid
)
# Check pipeline
ops = ds_grid_ndloc.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, '_perform_getitem')
self.assertEqual(
ops[1].args, [(slice(0, 2, None), slice(1, 3, None))]
)
self.assertEqual(ops[1].kwargs, {})
# Execute pipeline
self.assertEqual(
ds_grid_ndloc.pipeline(ds_grid_ndloc.dataset), ds_grid_ndloc
)
self.assertEqual(
ds_grid_ndloc.pipeline(self.ds2_grid), ds2_grid_ndloc
)
class SelectTestCase(DatasetPropertyTestCase):
def test_select_dataset(self):
ds_select = self.ds.select(b=10)
ds2_select = self.ds2.select(b=10)
self.assertNotEqual(ds_select, ds2_select)
# Dataset
self.assertEqual(
ds_select.dataset,
self.ds
)
# Check pipeline
ops = ds_select.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'select')
self.assertEqual(ops[1].args, [])
self.assertEqual(ops[1].kwargs, {'b': 10})
# Execute pipeline
self.assertEqual(ds_select.pipeline(ds_select.dataset), ds_select)
self.assertEqual(
ds_select.pipeline(self.ds2), ds2_select
)
def test_select_curve(self):
curve_select = self.ds.to.curve('a', 'b', groupby=[]).select(b=10)
curve2_select = self.ds2.to.curve('a', 'b', groupby=[]).select(b=10)
self.assertNotEqual(curve_select, curve2_select)
# Curve
self.assertEqual(
curve_select.dataset,
self.ds
)
# Check pipeline
ops = curve_select.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, 'select')
self.assertEqual(ops[2].args, [])
self.assertEqual(ops[2].kwargs, {'b': 10})
# Execute pipeline
self.assertEqual(
curve_select.pipeline(curve_select.dataset), curve_select
)
self.assertEqual(
curve_select.pipeline(self.ds2), curve2_select
)
class SortTestCase(DatasetPropertyTestCase):
def test_sort_curve(self):
curve_sorted = self.ds.to.curve('a', 'b', groupby=[]).sort('a')
curve_sorted2 = self.ds2.to.curve('a', 'b', groupby=[]).sort('a')
self.assertNotEqual(curve_sorted, curve_sorted2)
# Curve
self.assertEqual(
curve_sorted.dataset,
self.ds
)
# Check pipeline
ops = curve_sorted.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, 'sort')
self.assertEqual(ops[2].args, ['a'])
self.assertEqual(ops[2].kwargs, {})
# Execute pipeline
self.assertEqual(
curve_sorted.pipeline(curve_sorted.dataset), curve_sorted
)
self.assertEqual(
curve_sorted.pipeline(self.ds2), curve_sorted2
)
class SampleTestCase(DatasetPropertyTestCase):
def test_sample_curve(self):
curve_sampled = self.ds.to.curve('a', 'b', groupby=[]).sample([1, 2])
curve_sampled2 = self.ds2.to.curve('a', 'b', groupby=[]).sample([1, 2])
self.assertNotEqual(curve_sampled, curve_sampled2)
# Curve
self.assertEqual(
curve_sampled.dataset,
self.ds
)
# Check pipeline
ops = curve_sampled.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertEqual(ops[2].method_name, 'sample')
self.assertEqual(ops[2].args, [[1, 2]])
self.assertEqual(ops[2].kwargs, {})
# Execute pipeline
self.assertEqual(
curve_sampled.pipeline(curve_sampled.dataset), curve_sampled
)
self.assertEqual(
curve_sampled.pipeline(self.ds2), curve_sampled2
)
class ReduceTestCase(DatasetPropertyTestCase):
def test_reduce_dataset(self):
ds_reduced = self.ds.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).reduce('c', function=np.sum)
ds2_reduced = self.ds2.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).reduce('c', function=np.sum)
self.assertNotEqual(ds_reduced, ds2_reduced)
self.assertEqual(ds_reduced.dataset, self.ds)
self.assertEqual(ds2_reduced.dataset, self.ds2)
# Check pipeline
ops = ds_reduced.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'reindex')
self.assertEqual(ops[2].method_name, 'reduce')
self.assertEqual(ops[2].args, ['c'])
self.assertEqual(ops[2].kwargs, {'function': np.sum})
# Execute pipeline
self.assertEqual(ds_reduced.pipeline(ds_reduced.dataset), ds_reduced)
self.assertEqual(
ds_reduced.pipeline(self.ds2), ds2_reduced
)
class AggregateTestCase(DatasetPropertyTestCase):
def test_aggregate_dataset(self):
ds_aggregated = self.ds.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).aggregate('b', function=np.sum)
ds2_aggregated = self.ds2.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).aggregate('b', function=np.sum)
self.assertNotEqual(ds_aggregated, ds2_aggregated)
self.assertEqual(ds_aggregated.dataset, self.ds)
self.assertEqual(ds2_aggregated.dataset, self.ds2)
# Check pipeline
ops = ds_aggregated.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'reindex')
self.assertEqual(ops[2].method_name, 'aggregate')
self.assertEqual(ops[2].args, ['b'])
self.assertEqual(ops[2].kwargs, {'function': np.sum})
# Execute pipeline
self.assertEqual(
ds_aggregated.pipeline(ds_aggregated.dataset), ds_aggregated
)
self.assertEqual(
ds_aggregated.pipeline(self.ds2), ds2_aggregated
)
class GroupbyTestCase(DatasetPropertyTestCase):
def test_groupby_dataset(self):
ds_groups = self.ds.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).groupby('b')
ds2_groups = self.ds2.reindex(
kdims=['b', 'c'], vdims=['a', 'd']
).groupby('b')
self.assertNotEqual(ds_groups, ds2_groups)
for k in ds_groups.keys():
ds_group = ds_groups[k]
ds2_group = ds2_groups[k]
# Check pipeline
ops = ds_group.pipeline.operations
self.assertNotEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'reindex')
self.assertEqual(ops[2].method_name, 'groupby')
self.assertEqual(ops[2].args, ['b'])
self.assertEqual(ops[3].method_name, '__getitem__')
self.assertEqual(ops[3].args, [k])
# Execute pipeline
self.assertEqual(ds_group.pipeline(ds_group.dataset), ds_group)
self.assertEqual(
ds_group.pipeline(self.ds2), ds2_group
)
class AddDimensionTestCase(DatasetPropertyTestCase):
def test_add_dimension_dataset(self):
ds_dim_added = self.ds.add_dimension('new', 1, 17)
ds2_dim_added = self.ds2.add_dimension('new', 1, 17)
self.assertNotEqual(ds_dim_added, ds2_dim_added)
# Check dataset
self.assertEqual(ds_dim_added.dataset, self.ds)
self.assertEqual(ds2_dim_added.dataset, self.ds2)
# Check pipeline
ops = ds_dim_added.pipeline.operations
self.assertEqual(len(ops), 2)
self.assertIs(ops[0].output_type, Dataset)
self.assertEqual(ops[1].method_name, 'add_dimension')
self.assertEqual(ops[1].args, ['new', 1, 17])
self.assertEqual(ops[1].kwargs, {})
# Execute pipeline
self.assertEqual(
ds_dim_added.pipeline(ds_dim_added.dataset), ds_dim_added
)
self.assertEqual(
ds_dim_added.pipeline(self.ds2), ds2_dim_added,
)
# Add execute pipeline test for each method, using a different dataset (ds2)
#
class HistogramTestCase(DatasetPropertyTestCase):
def setUp(self):
super().setUp()
self.hist = self.ds.hist('a', adjoin=False, normed=False)
def test_construction(self):
self.assertEqual(self.hist.dataset, self.ds)
def test_clone(self):
self.assertEqual(self.hist.clone().dataset, self.ds)
def test_select_single(self):
sub_hist = self.hist.select(a=(1, None))
self.assertEqual(sub_hist.dataset, self.ds)
# Check pipeline
ops = sub_hist.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Apply)
self.assertEqual(ops[2].method_name, '__call__')
self.assertIsInstance(ops[2].args[0], histogram)
self.assertEqual(ops[3].method_name, 'select')
self.assertEqual(ops[3].args, [])
self.assertEqual(ops[3].kwargs, {'a': (1, None)})
# Execute pipeline
self.assertEqual(sub_hist.pipeline(sub_hist.dataset), sub_hist)
def test_select_multi(self):
# Add second selection on b. b is a dimension in hist.dataset but
# not in hist. Make sure that we only apply the a selection (and not
# the b selection) to the .dataset property
sub_hist = self.hist.select(a=(1, None), b=100)
self.assertNotEqual(
sub_hist.dataset,
self.ds.select(a=(1, None), b=100)
)
# Check dataset unchanged
self.assertEqual(
sub_hist.dataset,
self.ds
)
# Check pipeline
ops = sub_hist.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Apply)
self.assertEqual(ops[2].method_name, '__call__')
self.assertIsInstance(ops[2].args[0], histogram)
self.assertEqual(ops[3].method_name, 'select')
self.assertEqual(ops[3].args, [])
self.assertEqual(ops[3].kwargs, {'a': (1, None), 'b': 100})
# Execute pipeline
self.assertEqual(sub_hist.pipeline(sub_hist.dataset), sub_hist)
def test_hist_to_curve(self):
# No exception thrown
curve = self.hist.to.curve()
# Check pipeline
ops = curve.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Apply)
self.assertEqual(ops[2].method_name, '__call__')
self.assertIsInstance(ops[2].args[0], histogram)
self.assertIs(ops[3].output_type, Curve)
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
class DistributionTestCase(DatasetPropertyTestCase):
def setUp(self):
super().setUp()
self.distribution = self.ds.to(Distribution, kdims='a', groupby=[])
def test_distribution_dataset(self):
self.assertEqual(self.distribution.dataset, self.ds)
# Execute pipeline
self.assertEqual(
self.distribution.pipeline(self.distribution.dataset),
self.distribution,
)
class DatashaderTestCase(DatasetPropertyTestCase):
def setUp(self):
if None in (rasterize, datashade, dynspread):
raise SkipTest('Datashader could not be imported and cannot be tested.')
super().setUp()
def test_rasterize_curve(self):
img = rasterize(
self.ds.to(Curve, 'a', 'b', groupby=[]), dynamic=False
)
img2 = rasterize(
self.ds2.to(Curve, 'a', 'b', groupby=[]), dynamic=False
)
self.assertNotEqual(img, img2)
# Check dataset
self.assertEqual(img.dataset, self.ds)
# Check pipeline
ops = img.pipeline.operations
self.assertEqual(len(ops), 3)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertIsInstance(ops[2], rasterize)
# Execute pipeline
self.assertEqual(img.pipeline(img.dataset), img)
self.assertEqual(img.pipeline(self.ds2), img2)
def test_datashade_curve(self):
rgb = dynspread(datashade(
self.ds.to(Curve, 'a', 'b', groupby=[]), dynamic=False
), dynamic=False)
rgb2 = dynspread(datashade(
self.ds2.to(Curve, 'a', 'b', groupby=[]), dynamic=False
), dynamic=False)
self.assertNotEqual(rgb, rgb2)
# Check dataset
self.assertEqual(rgb.dataset, self.ds)
# Check pipeline
ops = rgb.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertIsInstance(ops[2], datashade)
self.assertIsInstance(ops[3], dynspread)
# Execute pipeline
self.assertEqual(rgb.pipeline(rgb.dataset), rgb)
self.assertEqual(rgb.pipeline(self.ds2), rgb2)
class AccessorTestCase(DatasetPropertyTestCase):
def test_apply_curve(self):
curve = self.ds.to.curve('a', 'b', groupby=[]).apply(
lambda c: Scatter(c.select(b=(20, None)).data)
)
curve2 = self.ds2.to.curve('a', 'b', groupby=[]).apply(
lambda c: Scatter(c.select(b=(20, None)).data)
)
self.assertNotEqual(curve, curve2)
# Check pipeline
ops = curve.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertIs(ops[2].output_type, Apply)
self.assertEqual(ops[2].kwargs, {'mode': None})
self.assertEqual(ops[3].method_name, '__call__')
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
self.assertEqual(
curve.pipeline(self.ds2), curve2
)
def test_redim_curve(self):
curve = self.ds.to.curve('a', 'b', groupby=[]).redim.unit(
a='kg', b='m'
)
curve2 = self.ds2.to.curve('a', 'b', groupby=[]).redim.unit(
a='kg', b='m'
)
self.assertNotEqual(curve, curve2)
# Check pipeline
ops = curve.pipeline.operations
self.assertEqual(len(ops), 4)
self.assertIs(ops[0].output_type, Dataset)
self.assertIs(ops[1].output_type, Curve)
self.assertIs(ops[2].output_type, Redim)
self.assertEqual(ops[2].kwargs, {'mode': 'dataset'})
self.assertEqual(ops[3].method_name, '__call__')
# Execute pipeline
self.assertEqual(curve.pipeline(curve.dataset), curve)
self.assertEqual(
curve.pipeline(self.ds2), curve2
)
class OperationTestCase(DatasetPropertyTestCase):
def test_propagate_dataset(self):
op = function.instance(
fn=lambda ds: ds.iloc[:5].clone(dataset=None, pipeline=None)
)
new_ds = op(self.ds)
self.assertEqual(new_ds.dataset, self.ds)
def test_do_not_propagate_dataset(self):
op = function.instance(
fn=lambda ds: ds.iloc[:5].clone(dataset=None, pipeline=None)
)
# Disable dataset propagation
op._propagate_dataset = False
new_ds = op(self.ds)
self.assertEqual(new_ds.dataset, new_ds)