File: C:/Users/fred/anaconda3/Lib/site-packages/holoviews/tests/plotting/plotly/test_callbacks.py
import uuid
from unittest import TestCase
from unittest.mock import Mock
import plotly.graph_objs as go
from holoviews import Tiles
from holoviews.plotting.plotly.callbacks import (
BoundsXCallback,
BoundsXYCallback,
BoundsYCallback,
RangeXCallback,
RangeXYCallback,
RangeYCallback,
Selection1DCallback,
)
from holoviews.streams import (
BoundsX,
BoundsXY,
BoundsY,
RangeX,
RangeXY,
RangeY,
Selection1D,
)
def mock_plot(trace_uid=None):
# Build a mock to stand in for a PlotlyPlot subclass
if trace_uid is None:
trace_uid = str(uuid.uuid4())
plot = Mock()
plot.trace_uid = trace_uid
return plot
def build_callback_set(callback_cls, trace_uids, stream_type, num_streams=2):
"""
Build a collection of plots, callbacks, and streams for a given callback class and
a list of trace_uids
"""
plots = []
streamss = []
callbacks = []
eventss = []
for trace_uid in trace_uids:
plot = mock_plot(trace_uid)
streams, event_list = [], []
for _ in range(num_streams):
events = []
stream = stream_type()
def cb(events=events, **kwargs):
events.append(kwargs)
stream.add_subscriber(cb)
streams.append(stream)
event_list.append(events)
callback = callback_cls(plot, streams, None)
plots.append(plot)
streamss.append(streams)
callbacks.append(callback)
eventss.append(event_list)
return plots, streamss, callbacks, eventss
class TestCallbacks(TestCase):
def setUp(self):
self.fig_dict = go.Figure({
'data': [
{'type': 'scatter',
'y': [1, 2, 3],
'uid': 'first'},
{'type': 'bar',
'y': [1, 2, 3],
'uid': 'second',
'xaxis': 'x',
'yaxis': 'y'},
{'type': 'scatter',
'y': [1, 2, 3],
'uid': 'third',
'xaxis': 'x2',
'yaxis': 'y2'},
{'type': 'bar',
'y': [1, 2, 3],
'uid': 'forth',
'xaxis': 'x3',
'yaxis': 'y3'},
],
'layout': {
'title': {'text': 'Figure Title'}}
}).to_dict()
self.mapbox_fig_dict = go.Figure({
'data': [
{'type': 'scattermapbox', 'uid': 'first', 'subplot': 'mapbox'},
{'type': 'scattermapbox', 'uid': 'second', 'subplot': 'mapbox2'},
{'type': 'scattermapbox', 'uid': 'third', 'subplot': 'mapbox3'}
],
'layout': {
'title': {'text': 'Figure Title'},
}
}).to_dict()
# Precompute a pair of lat/lon, easting/northing, mapbox coord values
self.lon_range1, self.lat_range1 = (10, 30), (20, 40)
self.easting_range1, self.northing_range1 = Tiles.lon_lat_to_easting_northing(
self.lon_range1, self.lat_range1
)
self.easting_range1 = tuple(self.easting_range1)
self.northing_range1 = tuple(self.northing_range1)
self.mapbox_coords1 = [
[self.lon_range1[0], self.lat_range1[1]],
[self.lon_range1[1], self.lat_range1[1]],
[self.lon_range1[1], self.lat_range1[0]],
[self.lon_range1[0], self.lat_range1[0]]
]
self.lon_range2, self.lat_range2 = (-50, -30), (-70, -40)
self.easting_range2, self.northing_range2 = Tiles.lon_lat_to_easting_northing(
self.lon_range2, self.lat_range2
)
self.easting_range2 = tuple(self.easting_range2)
self.northing_range2 = tuple(self.northing_range2)
self.mapbox_coords2 = [
[self.lon_range2[0], self.lat_range2[1]],
[self.lon_range2[1], self.lat_range2[1]],
[self.lon_range2[1], self.lat_range2[0]],
[self.lon_range2[0], self.lat_range2[0]]
]
def testCallbackClassInstanceTracking(self):
# Each callback class should track all active instances of its own class in a
# weak value dictionary. Here we make sure that instances stay separated per
# class
plot1 = mock_plot()
plot2 = mock_plot()
plot3 = mock_plot()
# Check RangeXYCallback
rangexy_cb = RangeXYCallback(plot1, [], None)
self.assertIn(plot1.trace_uid, RangeXYCallback.instances)
self.assertIs(rangexy_cb, RangeXYCallback.instances[plot1.trace_uid])
# Check BoundsXYCallback
boundsxy_cb = BoundsXYCallback(plot2, [], None)
self.assertIn(plot2.trace_uid, BoundsXYCallback.instances)
self.assertIs(boundsxy_cb, BoundsXYCallback.instances[plot2.trace_uid])
# Check Selection1DCallback
selection1d_cb = Selection1DCallback(plot3, [], None)
self.assertIn(plot3.trace_uid, Selection1DCallback.instances)
self.assertIs(selection1d_cb, Selection1DCallback.instances[plot3.trace_uid])
# Check that objects don't show up as instances in the wrong class
self.assertNotIn(plot1.trace_uid, BoundsXYCallback.instances)
self.assertNotIn(plot1.trace_uid, Selection1DCallback.instances)
self.assertNotIn(plot2.trace_uid, RangeXYCallback.instances)
self.assertNotIn(plot2.trace_uid, Selection1DCallback.instances)
self.assertNotIn(plot3.trace_uid, RangeXYCallback.instances)
self.assertNotIn(plot3.trace_uid, BoundsXYCallback.instances)
def testRangeXYCallbackEventData(self):
for viewport in [
{'xaxis.range': [1, 4], 'yaxis.range': [-1, 5]},
{'xaxis.range[0]': 1, 'xaxis.range[1]': 4,
'yaxis.range[0]': -1, 'yaxis.range[1]': 5},
]:
event_data = RangeXYCallback.get_event_data_from_property_update(
"viewport", viewport, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'x_range': (1, 4), 'y_range': (-1, 5)},
'second': {'x_range': (1, 4), 'y_range': (-1, 5)},
})
def testRangeXCallbackEventData(self):
for viewport in [
{'xaxis.range': [1, 4], 'yaxis.range': [-1, 5]},
{'xaxis.range[0]': 1, 'xaxis.range[1]': 4,
'yaxis.range[0]': -1, 'yaxis.range[1]': 5},
]:
event_data = RangeXCallback.get_event_data_from_property_update(
"viewport", viewport, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'x_range': (1, 4)},
'second': {'x_range': (1, 4)},
})
def testRangeYCallbackEventData(self):
for viewport in [
{'xaxis.range': [1, 4], 'yaxis.range': [-1, 5]},
{'xaxis.range[0]': 1, 'xaxis.range[1]': 4,
'yaxis.range[0]': -1, 'yaxis.range[1]': 5},
]:
event_data = RangeYCallback.get_event_data_from_property_update(
"viewport", viewport, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'y_range': (-1, 5)},
'second': {'y_range': (-1, 5)},
})
def testMapboxRangeXYCallbackEventData(self):
relayout_data = {
'mapbox._derived': {"coordinates": self.mapbox_coords1},
'mapbox3._derived': {"coordinates": self.mapbox_coords2}
}
event_data = RangeXYCallback.get_event_data_from_property_update(
"relayout_data", relayout_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'x_range': self.easting_range1, 'y_range': self.northing_range1},
'third': {'x_range': self.easting_range2, 'y_range': self.northing_range2},
})
def testMapboxRangeXCallbackEventData(self):
relayout_data = {
'mapbox._derived': {"coordinates": self.mapbox_coords1},
'mapbox3._derived': {"coordinates": self.mapbox_coords2}
}
event_data = RangeXCallback.get_event_data_from_property_update(
"relayout_data", relayout_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'x_range': self.easting_range1},
'third': {'x_range': self.easting_range2},
})
def testMapboxRangeYCallbackEventData(self):
relayout_data = {
'mapbox._derived': {"coordinates": self.mapbox_coords1},
'mapbox3._derived': {"coordinates": self.mapbox_coords2}
}
event_data = RangeYCallback.get_event_data_from_property_update(
"relayout_data", relayout_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'y_range': self.northing_range1},
'third': {'y_range': self.northing_range2},
})
def testRangeCallbacks(self):
# Build callbacks
range_classes = [RangeXYCallback, RangeXCallback, RangeYCallback]
xyplots, xystreamss, xycallbacks, xyevents = build_callback_set(
RangeXYCallback, ['first', 'second', 'third', 'forth', 'other'],
RangeXY, 2
)
xplots, xstreamss, xcallbacks, xevents = build_callback_set(
RangeXCallback, ['first', 'second', 'third', 'forth', 'other'],
RangeX, 2
)
yplots, ystreamss, ycallbacks, yevents = build_callback_set(
RangeYCallback, ['first', 'second', 'third', 'forth', 'other'],
RangeY, 2
)
# Sanity check the length of the streams lists
for xystreams in xystreamss:
self.assertEqual(len(xystreams), 2)
# Change viewport on first set of axes
viewport1 = {'xaxis.range': [1, 4], 'yaxis.range': [-1, 5]}
for cb_cls in range_classes:
cb_cls.update_streams_from_property_update(
"viewport", viewport1, self.fig_dict
)
# Check that all streams attached to 'first' and 'second' plots were triggered
for xystream, xstream, ystream in zip(
xystreamss[0] + xystreamss[1],
xstreamss[0] + xstreamss[1],
ystreamss[0] + ystreamss[1],
):
assert xystream.x_range == (1, 4)
assert xystream.y_range == (-1, 5)
assert xstream.x_range == (1, 4)
assert ystream.y_range == (-1, 5)
# And that no other streams were triggered
for xystream, xstream, ystream in zip(
xystreamss[2] + xystreamss[3],
xstreamss[2] + xstreamss[3],
ystreamss[2] + ystreamss[3],
):
assert xystream.x_range is None
assert xystream.y_range is None
assert xstream.x_range is None
assert ystream.y_range is None
# Change viewport on second set of axes
viewport2 = {'xaxis2.range': [2, 5], 'yaxis2.range': [0, 6]}
for cb_cls in range_classes:
cb_cls.update_streams_from_property_update(
"viewport", viewport2, self.fig_dict
)
# Check that all streams attached to 'third' were triggered
for xystream, xstream, ystream in zip(
xystreamss[2], xstreamss[2], ystreamss[2]
):
assert xystream.x_range == (2, 5)
assert xystream.y_range == (0, 6)
assert xstream.x_range == (2, 5)
assert ystream.y_range == (0, 6)
# Change viewport on third set of axes
viewport3 = {'xaxis3.range': [3, 6], 'yaxis3.range': [1, 7]}
for cb_cls in range_classes:
cb_cls.update_streams_from_property_update(
"viewport", viewport3, self.fig_dict
)
# Check that all streams attached to 'forth' were triggered
for xystream, xstream, ystream in zip(
xystreamss[3], xstreamss[3], ystreamss[3]
):
assert xystream.x_range == (3, 6)
assert xystream.y_range == (1, 7)
assert xstream.x_range == (3, 6)
assert ystream.y_range == (1, 7)
# Check that streams attached to a trace not in this plot are not triggered
for xyevent, xevent, yevent in zip(
xyevents[4], xevents[4], yevents[4]
):
assert len(xyevent) == 0
assert len(xevent) == 0
assert len(yevent) == 0
def testBoundsXYCallbackEventData(self):
selected_data1 = {'range': {'x': [1, 4], 'y': [-1, 5]}}
event_data = BoundsXYCallback.get_event_data_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'bounds': (1, -1, 4, 5)},
'second': {'bounds': (1, -1, 4, 5)},
'third': {'bounds': None},
'forth': {'bounds': None}
})
def testBoundsXCallbackEventData(self):
selected_data1 = {'range': {'x': [1, 4], 'y': [-1, 5]}}
event_data = BoundsXCallback.get_event_data_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'boundsx': (1, 4)},
'second': {'boundsx': (1, 4)},
'third': {'boundsx': None},
'forth': {'boundsx': None}
})
def testBoundsYCallbackEventData(self):
selected_data1 = {'range': {'x': [1, 4], 'y': [-1, 5]}}
event_data = BoundsYCallback.get_event_data_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'boundsy': (-1, 5)},
'second': {'boundsy': (-1, 5)},
'third': {'boundsy': None},
'forth': {'boundsy': None}
})
def testMapboxBoundsXYCallbackEventData(self):
selected_data = {"range": {'mapbox2': [
[self.lon_range1[0], self.lat_range1[0]],
[self.lon_range1[1], self.lat_range1[1]]
]}}
event_data = BoundsXYCallback.get_event_data_from_property_update(
"selected_data", selected_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'bounds': None},
'second': {'bounds': (
self.easting_range1[0], self.northing_range1[0],
self.easting_range1[1], self.northing_range1[1]
)},
'third': {'bounds': None}
})
def testMapboxBoundsXCallbackEventData(self):
selected_data = {"range": {'mapbox': [
[self.lon_range1[0], self.lat_range1[0]],
[self.lon_range1[1], self.lat_range1[1]]
]}}
event_data = BoundsXCallback.get_event_data_from_property_update(
"selected_data", selected_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'boundsx': (
self.easting_range1[0], self.easting_range1[1],
)},
'second': {'boundsx': None},
'third': {'boundsx': None}
})
def testMapboxBoundsYCallbackEventData(self):
selected_data = {"range": {'mapbox3': [
[self.lon_range1[0], self.lat_range1[0]],
[self.lon_range1[1], self.lat_range1[1]]
]}}
event_data = BoundsYCallback.get_event_data_from_property_update(
"selected_data", selected_data, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'boundsy': None},
'second': {'boundsy': None},
'third': {'boundsy': (
self.northing_range1[0], self.northing_range1[1]
)},
})
def testBoundsCallbacks(self):
# Build callbacks
bounds_classes = [BoundsXYCallback, BoundsXCallback, BoundsYCallback]
xyplots, xystreamss, xycallbacks, xyevents = build_callback_set(
BoundsXYCallback, ['first', 'second', 'third', 'forth', 'other'],
BoundsXY, 2
)
xplots, xstreamss, xcallbacks, xevents = build_callback_set(
BoundsXCallback, ['first', 'second', 'third', 'forth', 'other'],
BoundsX, 2
)
yplots, ystreamss, ycallbacks, yevents = build_callback_set(
BoundsYCallback, ['first', 'second', 'third', 'forth', 'other'],
BoundsY, 2
)
# box selection on first set of axes
selected_data1 = {'range': {'x': [1, 4], 'y': [-1, 5]}}
for cb_cls in bounds_classes:
cb_cls.update_streams_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
# Check that all streams attached to 'first' and 'second' plots were triggered
for xystream, xstream, ystream in zip(
xystreamss[0] + xystreamss[1],
xstreamss[0] + xstreamss[1],
ystreamss[0] + ystreamss[1],
):
assert xystream.bounds == (1, -1, 4, 5)
assert xstream.boundsx == (1, 4)
assert ystream.boundsy == (-1, 5)
# Check that streams attached to plots in other subplots are called with None
# to clear their bounds
for xystream, xstream, ystream in zip(
xystreamss[2] + xystreamss[3],
xstreamss[2] + xstreamss[3],
ystreamss[2] + ystreamss[3],
):
assert xystream.bounds is None
assert xstream.boundsx is None
assert ystream.boundsy is None
# box select on second set of axes
selected_data2 = {'range': {'x2': [2, 5], 'y2': [0, 6]}}
for cb_cls in bounds_classes:
cb_cls.update_streams_from_property_update(
"selected_data", selected_data2, self.fig_dict
)
# Check that all streams attached to 'second' were triggered
for xystream, xstream, ystream in zip(
xystreamss[2], xstreamss[2], ystreamss[2],
):
assert xystream.bounds == (2, 0, 5, 6)
assert xstream.boundsx == (2, 5)
assert ystream.boundsy == (0, 6)
# box select on third set of axes
selected_data3 = {'range': {'x3': [3, 6], 'y3': [1, 7]}}
for cb_cls in bounds_classes:
cb_cls.update_streams_from_property_update(
"selected_data", selected_data3, self.fig_dict
)
# Check that all streams attached to 'third' were triggered
for xystream, xstream, ystream in zip(
xystreamss[3], xstreamss[3], ystreamss[3],
):
assert xystream.bounds == (3, 1, 6, 7)
assert xstream.boundsx == (3, 6)
assert ystream.boundsy == (1, 7)
# lasso select on first set of axes should clear all bounds
selected_data_lasso = {'lassoPoints': {'x': [1, 4, 2], 'y': [-1, 5, 2]}}
for cb_cls in bounds_classes:
cb_cls.update_streams_from_property_update(
"selected_data", selected_data_lasso, self.fig_dict
)
# Check that all streams attached to this figure are called with None
# to clear their bounds
for xystream, xstream, ystream in zip(
xystreamss[0] + xystreamss[1] + xystreamss[2] + xystreamss[3],
xstreamss[0] + xstreamss[1] + xstreamss[2] + xstreamss[3],
ystreamss[0] + ystreamss[1] + ystreamss[2] + ystreamss[3],
):
assert xystream.bounds is None
assert xstream.boundsx is None
assert ystream.boundsy is None
# Check that streams attached to plots not in this figure are not called
for xyevent, xevent, yevent in zip(
xyevents[4], xevents[4], yevents[4]
):
assert xyevent == []
assert xevent == []
assert yevent == []
def testSelection1DCallbackEventData(self):
selected_data1 = {'points': [
{"pointNumber": 0, "curveNumber": 0},
{"pointNumber": 2, "curveNumber": 0},
]}
event_data = Selection1DCallback.get_event_data_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
self.assertEqual(event_data, {
'first': {'index': [0, 2]},
'second': {'index': []},
'third': {'index': []},
'forth': {'index': []}
})
def testMapboxSelection1DCallbackEventData(self):
selected_data1 = {'points': [
{"pointNumber": 0, "curveNumber": 1},
{"pointNumber": 2, "curveNumber": 1},
]}
event_data = Selection1DCallback.get_event_data_from_property_update(
"selected_data", selected_data1, self.mapbox_fig_dict
)
self.assertEqual(event_data, {
'first': {'index': []},
'second': {'index': [0, 2]},
'third': {'index': []},
})
def testSelection1DCallback(self):
plots, streamss, callbacks, sel_events = build_callback_set(
Selection1DCallback, ['first', 'second', 'third', 'forth', 'other'],
Selection1D, 2
)
# Select points from the 'first' plot (first set of axes)
selected_data1 = {'points': [
{"pointNumber": 0, "curveNumber": 0},
{"pointNumber": 2, "curveNumber": 0},
]}
Selection1DCallback.update_streams_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
# Check that all streams attached to the 'first' plots were triggered
for stream, events in zip(streamss[0], sel_events[0]):
assert stream.index == [0, 2]
assert len(events) == 1
# Check that all streams attached to other plots in this figure were triggered
# with empty selection
for stream in streamss[1] + streamss[2] + streamss[3]:
assert stream.index == []
# Select points from the 'first' and 'second' plot (first set of axes)
selected_data1 = {'points': [
{"pointNumber": 0, "curveNumber": 0},
{"pointNumber": 1, "curveNumber": 0},
{"pointNumber": 1, "curveNumber": 1},
{"pointNumber": 2, "curveNumber": 1},
]}
Selection1DCallback.update_streams_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
# Check that all streams attached to the 'first' plot were triggered
for stream in streamss[0]:
assert stream.index == [0, 1]
# Check that all streams attached to the 'second' plot were triggered
for stream in streamss[1]:
assert stream.index == [1, 2]
# Check that all streams attached to other plots in this figure were triggered
# with empty selection
for stream in streamss[2] + streamss[3]:
assert stream.index == []
# Select points from the 'forth' plot (third set of axes)
selected_data1 = {'points': [
{"pointNumber": 0, "curveNumber": 3},
{"pointNumber": 2, "curveNumber": 3},
]}
Selection1DCallback.update_streams_from_property_update(
"selected_data", selected_data1, self.fig_dict
)
# Check that all streams attached to the 'forth' plot were triggered
for stream, _events in zip(streamss[3], sel_events[3]):
assert stream.index == [0, 2]
# Check that streams attached to plots not in this figure are not called
for _stream, events in zip(streamss[4], sel_events[4]):
assert len(events) == 0