Source code for holoviews.plotting.plotly.callbacks

from weakref import WeakValueDictionary

from param.parameterized import add_metaclass

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

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(PlotlyCallbackMetaClass, cls).__init__(name, bases, attrs)

        # Create weak-value dictionary to hold instances of the class
        cls.instances = WeakValueDictionary()

    def __call__(cls, *args, **kwargs):
        inst = super(PlotlyCallbackMetaClass, cls).__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 colleciton when this
        # happens.
        cls.instances[inst.plot.trace_uid] = inst

        return inst


@add_metaclass(PlotlyCallbackMetaClass)
class PlotlyCallback(object):

    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_value, fig_dict):
        event_data = cls.get_event_data_from_property_update(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_value, fig_dict):
        raise NotImplementedError


class Selection1DCallback(PlotlyCallback):
    callback_property = "selected_data"

    @classmethod
    def get_event_data_from_property_update(cls, 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_property = "selected_data"
    boundsx = False
    boundsy = False

    @classmethod
    def get_event_data_from_property_update(cls, selected_data, fig_dict):
        traces = fig_dict.get('data', [])

        if not selected_data or 'range' not in selected_data:
            # No valid box selection
            box = None
        else:
            # Get x and y axis references
            box = selected_data["range"]
            axis_refs = list(box)
            xref = [ref for ref in axis_refs if ref.startswith('x')][0]
            yref = [ref for ref in axis_refs if ref.startswith('y')][0]

        # Process traces
        event_data = {}
        for trace_ind, trace in enumerate(traces):
            trace_type = trace.get('type', 'scatter')
            trace_uid = trace.get('uid', None)

            if _trace_to_subplot.get(trace_type, None) != ['xaxis', 'yaxis']:
                continue

            if (box and trace.get('xaxis', 'x') == xref and
                    trace.get('yaxis', 'y') == yref):

                new_bounds = (box[xref][0], box[yref][0], box[xref][1], box[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 = dict()

                event_data[trace_uid] = stream_data
            else:
                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 = dict()

                event_data[trace_uid] = stream_data

        return event_data


class BoundsXYCallback(BoundsCallback):
    boundsx = True
    boundsy = True


class BoundsXCallback(BoundsCallback):
    boundsx = True


class BoundsYCallback(BoundsCallback):
    boundsy = True


class RangeCallback(PlotlyCallback):
    callback_property = "viewport"
    x_range = False
    y_range = False

    @classmethod
    def get_event_data_from_property_update(cls, viewport, fig_dict):

        traces = fig_dict.get('data', [])

        # Process traces
        event_data = {}
        for trace_ind, trace in enumerate(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 = '{xaxis}.range'.format(xaxis=xaxis)
            yprop = '{yaxis}.range'.format(yaxis=yaxis)

            if not viewport:
                x_range = None
                y_range = None
            elif xprop in viewport and yprop in viewport:
                x_range = tuple(viewport[xprop])
                y_range = tuple(viewport[yprop])
            elif xprop + "[0]" in viewport and xprop + "[1]" in viewport and \
                    yprop + "[0]" in viewport and yprop + "[1]" in viewport :
                x_range = (viewport[xprop + "[0]"], viewport[xprop + "[1]"])
                y_range = (viewport[yprop + "[0]"], viewport[yprop + "[1]"])
            else:
                x_range = None
                y_range = None

            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