Source code for holoviews.plotting.plotly.callbacks

from weakref import WeakValueDictionary

from ...element import Tiles
from ...streams import (
    BoundsX,
    BoundsXY,
    BoundsY,
    RangeX,
    RangeXY,
    RangeY,
    Selection1D,
    SelectionXY,
    Stream,
)
from .util import _trace_to_subplot


[docs]class PlotlyCallbackMetaClass(type): """ Metaclass for PlotlyCallback classes. We want each callback class to keep track of all of the instances of the class. Using a meta class here lets us keep the logic for instance tracking in one place. """ def __init__(cls, name, bases, attrs): super().__init__(name, bases, attrs) # Create weak-value dictionary to hold instances of the class cls.instances = WeakValueDictionary() def __call__(cls, *args, **kwargs): inst = super().__call__(*args, **kwargs) # Store weak reference to the callback instance in the _instances # WeakValueDictionary. This will allow instances to be garbage collected and # the references will be automatically removed from the collection when this # happens. cls.instances[inst.plot.trace_uid] = inst return inst
class PlotlyCallback(metaclass=PlotlyCallbackMetaClass): def __init__(self, plot, streams, source, **params): self.plot = plot self.streams = streams self.source = source self.last_event = None @classmethod def update_streams_from_property_update(cls, property, property_value, fig_dict): event_data = cls.get_event_data_from_property_update( property, property_value, fig_dict ) streams = [] for trace_uid, stream_data in event_data.items(): if trace_uid in cls.instances: cb = cls.instances[trace_uid] try: unchanged = stream_data == cb.last_event except Exception: unchanged = False if unchanged: continue cb.last_event = stream_data for stream in cb.streams: stream.update(**stream_data) streams.append(stream) try: Stream.trigger(streams) except Exception as e: raise e @classmethod def get_event_data_from_property_update(cls, property, property_value, fig_dict): raise NotImplementedError class Selection1DCallback(PlotlyCallback): callback_properties = ["selected_data"] @classmethod def get_event_data_from_property_update(cls, property, selected_data, fig_dict): traces = fig_dict.get('data', []) # build event data and compute which trace UIDs are eligible # Look up callback with UID # graph reference and update the streams point_inds = {} if selected_data: for point in selected_data['points']: point_inds.setdefault(point['curveNumber'], []) point_inds[point['curveNumber']].append(point['pointNumber']) event_data = {} for trace_ind, trace in enumerate(traces): trace_uid = trace.get('uid', None) new_index = point_inds.get(trace_ind, []) event_data[trace_uid] = dict(index=new_index) return event_data class BoundsCallback(PlotlyCallback): callback_properties = ["selected_data"] boundsx = False boundsy = False @classmethod def get_event_data_from_property_update(cls, property, selected_data, fig_dict): traces = fig_dict.get('data', []) # Initialize event data by clearing box selection on everything event_data = {} for trace in traces: trace_uid = trace.get('uid', None) if cls.boundsx and cls.boundsy: stream_data = dict(bounds=None) elif cls.boundsx: stream_data = dict(boundsx=None) elif cls.boundsy: stream_data = dict(boundsy=None) else: stream_data = {} event_data[trace_uid] = stream_data range_data = (selected_data or {}).get("range", {}) cls.update_event_data_xyaxis(range_data, traces, event_data) cls.update_event_data_mapbox(range_data, traces, event_data) return event_data @classmethod def update_event_data_xyaxis(cls, range_data, traces, event_data): # Process traces for trace in traces: trace_type = trace.get('type', 'scatter') trace_uid = trace.get('uid', None) if _trace_to_subplot.get(trace_type, None) != ['xaxis', 'yaxis']: continue xref = trace.get('xaxis', 'x') yref = trace.get('yaxis', 'y') if xref in range_data and yref in range_data: new_bounds = ( range_data[xref][0], range_data[yref][0], range_data[xref][1], range_data[yref][1] ) if cls.boundsx and cls.boundsy: stream_data = dict(bounds=new_bounds) elif cls.boundsx: stream_data = dict(boundsx=(new_bounds[0], new_bounds[2])) elif cls.boundsy: stream_data = dict(boundsy=(new_bounds[1], new_bounds[3])) else: stream_data = {} event_data[trace_uid] = stream_data @classmethod def update_event_data_mapbox(cls, range_data, traces, event_data): # Process traces for trace in traces: trace_type = trace.get('type', 'scatter') trace_uid = trace.get('uid', None) if _trace_to_subplot.get(trace_type, None) != ['mapbox']: continue mapbox_ref = trace.get('subplot', 'mapbox') if mapbox_ref in range_data: lon_bounds = [range_data[mapbox_ref][0][0], range_data[mapbox_ref][1][0]] lat_bounds = [range_data[mapbox_ref][0][1], range_data[mapbox_ref][1][1]] easting, northing = Tiles.lon_lat_to_easting_northing(lon_bounds, lat_bounds) new_bounds = (easting[0], northing[0], easting[1], northing[1]) if cls.boundsx and cls.boundsy: stream_data = dict(bounds=new_bounds) elif cls.boundsx: stream_data = dict(boundsx=(new_bounds[0], new_bounds[2])) elif cls.boundsy: stream_data = dict(boundsy=(new_bounds[1], new_bounds[3])) else: stream_data = {} event_data[trace_uid] = stream_data class BoundsXYCallback(BoundsCallback): boundsx = True boundsy = True class BoundsXCallback(BoundsCallback): boundsx = True class BoundsYCallback(BoundsCallback): boundsy = True class RangeCallback(PlotlyCallback): callback_properties = ["viewport", "relayout_data"] x_range = False y_range = False @classmethod def get_event_data_from_property_update(cls, property, property_value, fig_dict): traces = fig_dict.get('data', []) if property == "viewport": event_data = cls.build_event_data_from_viewport(traces, property_value) else: event_data = cls.build_event_data_from_relayout_data(traces, property_value) return event_data @classmethod def build_event_data_from_viewport(cls, traces, property_value): # Process traces event_data = {} for trace in traces: trace_type = trace.get('type', 'scatter') trace_uid = trace.get('uid', None) if _trace_to_subplot.get(trace_type, None) != ['xaxis', 'yaxis']: continue xaxis = trace.get('xaxis', 'x').replace('x', 'xaxis') yaxis = trace.get('yaxis', 'y').replace('y', 'yaxis') xprop = f'{xaxis}.range' yprop = f'{yaxis}.range' if not property_value: x_range = None y_range = None elif xprop in property_value and yprop in property_value: x_range = tuple(property_value[xprop]) y_range = tuple(property_value[yprop]) elif xprop + "[0]" in property_value and xprop + "[1]" in property_value and \ yprop + "[0]" in property_value and yprop + "[1]" in property_value: x_range = (property_value[xprop + "[0]"],property_value[xprop + "[1]"]) y_range = (property_value[yprop + "[0]"], property_value[yprop + "[1]"]) else: continue stream_data = {} if cls.x_range: stream_data['x_range'] = x_range if cls.y_range: stream_data['y_range'] = y_range event_data[trace_uid] = stream_data return event_data @classmethod def build_event_data_from_relayout_data(cls, traces, property_value): # Process traces event_data = {} for trace in traces: trace_type = trace.get('type', 'scattermapbox') trace_uid = trace.get('uid', None) if _trace_to_subplot.get(trace_type, None) != ['mapbox']: continue subplot_id = trace.get("subplot", "mapbox") derived_prop = subplot_id + "._derived" if not property_value: x_range = None y_range = None elif "coordinates" in property_value.get(derived_prop, {}): coords = property_value[derived_prop]["coordinates"] ((lon_top_left, lat_top_left), (lon_top_right, lat_top_right), (lon_bottom_right, lat_bottom_right), (lon_bottom_left, lat_bottom_left)) = coords lon_left = min(lon_top_left, lon_bottom_left) lon_right = max(lon_top_right, lon_bottom_right) lat_bottom = min(lat_bottom_left, lat_bottom_right) lat_top = max(lat_top_left, lat_top_right) x_range, y_range = Tiles.lon_lat_to_easting_northing( [lon_left, lon_right], [lat_bottom, lat_top] ) x_range = tuple(x_range) y_range = tuple(y_range) else: continue stream_data = {} if cls.x_range: stream_data['x_range'] = x_range if cls.y_range: stream_data['y_range'] = y_range event_data[trace_uid] = stream_data return event_data class RangeXYCallback(RangeCallback): x_range = True y_range = True class RangeXCallback(RangeCallback): x_range = True class RangeYCallback(RangeCallback): y_range = True callbacks = Stream._callbacks['plotly'] callbacks[Selection1D] = Selection1DCallback callbacks[SelectionXY] = BoundsXYCallback callbacks[BoundsXY] = BoundsXYCallback callbacks[BoundsX] = BoundsXCallback callbacks[BoundsY] = BoundsYCallback callbacks[RangeXY] = RangeXYCallback callbacks[RangeX] = RangeXCallback callbacks[RangeY] = RangeYCallback