Merge pull request #951 from ioam/dynamic_callable

Dynamic Callable API

holoviews/core/operation.py

@@ -14,7 +14,7 @@
 from .element import Element, HoloMap, GridSpace, Collator
 from .layout import Layout
 from .overlay import NdOverlay, Overlay
-from .spaces import DynamicMap
+from .spaces import DynamicMap, Callable
 from .traversal import unique_dimkeys
 from . import util
 
@@ -160,6 +160,16 @@ def __call__(self, element, **params):
         return processed
 
 
+class OperationCallable(Callable):
+    """
+    OperationCallable allows wrapping an ElementOperation and the
+    objects it is processing to allow traversing the operations
+    applied on a DynamicMap.
+    """
+
+    operation = param.ClassSelector(class_=ElementOperation, doc="""
+        The ElementOperation being wrapped.""")
+
 
 class MapOperation(param.ParameterizedFunction):
     """

holoviews/core/overlay.py

@@ -24,10 +24,14 @@ class Overlayable(object):
 
     def __mul__(self, other):
         if type(other).__name__ == 'DynamicMap':
-            from ..util import Dynamic
-            def dynamic_mul(element):
+            from .spaces import Callable
+            def dynamic_mul(*args, **kwargs):
+                element = other[args]
                 return self * element
-            return Dynamic(other, operation=dynamic_mul)
+            callback = Callable(callable_function=dynamic_mul,
+                                inputs=[self, other])
+            return other.clone(shared_data=False, callback=callback,
+                               streams=[])
         if isinstance(other, UniformNdMapping) and not isinstance(other, CompositeOverlay):
             items = [(k, self * v) for (k, v) in other.items()]
             return other.clone(items)

holoviews/core/spaces.py

@@ -120,12 +120,13 @@ def _dynamic_mul(self, dimensions, other, keys):
             map_obj = self if isinstance(self, DynamicMap) else other
             mode = map_obj.mode
 
-        def dynamic_mul(*key):
+        def dynamic_mul(*key, **kwargs):
             key = key[0] if mode == 'open' else key
             layers = []
             try:
                 if isinstance(self, DynamicMap):
-                    _, self_el = util.get_dynamic_item(self, dimensions, key)
+                    safe_key = () if not self.kdims else key
+                    _, self_el = util.get_dynamic_item(self, dimensions, safe_key)
                     if self_el is not None:
                         layers.append(self_el)
                 else:
@@ -134,19 +135,21 @@ def dynamic_mul(*key):
                 pass
             try:
                 if isinstance(other, DynamicMap):
-                    _, other_el = util.get_dynamic_item(other, dimensions, key)
+                    safe_key = () if not other.kdims else key
+                    _, other_el = util.get_dynamic_item(other, dimensions, safe_key)
                     if other_el is not None:
                         layers.append(other_el)
                 else:
                     layers.append(other[key])
             except KeyError:
                 pass
             return Overlay(layers)
+        callback = Callable(callable_function=dynamic_mul, inputs=[self, other])
         if map_obj:
-            return map_obj.clone(callback=dynamic_mul, shared_data=False,
-                                 kdims=dimensions)
+            return map_obj.clone(callback=callback, shared_data=False,
+                                 kdims=dimensions, streams=[])
         else:
-            return DynamicMap(callback=dynamic_mul, kdims=dimensions)
+            return DynamicMap(callback=callback, kdims=dimensions)
 
 
     def __mul__(self, other):
@@ -204,10 +207,13 @@ def __mul__(self, other):
             return self.clone(items, kdims=dimensions, label=self._label, group=self._group)
         elif isinstance(other, self.data_type):
             if isinstance(self, DynamicMap):
-                from ..util import Dynamic
-                def dynamic_mul(element):
+                def dynamic_mul(*args, **kwargs):
+                    element = self[args]
                     return element * other
-                return Dynamic(self, operation=dynamic_mul)
+                callback = Callable(callable_function=dynamic_mul,
+                                    inputs=[self, other])
+                return self.clone(shared_data=False, callback=callback,
+                                  streams=[])
             items = [(k, v * other) for (k, v) in self.data.items()]
             return self.clone(items, label=self._label, group=self._group)
         else:
@@ -393,6 +399,38 @@ def hist(self, num_bins=20, bin_range=None, adjoin=True, individually=True, **kw
                 return histmaps[0]
 
 
+class Callable(param.Parameterized):
+    """
+    Callable allows wrapping callbacks on one or more DynamicMaps
+    allowing their inputs (and in future outputs) to be defined.
+    This makes it possible to wrap DynamicMaps with streams and
+    makes it possible to traverse the graph of operations applied
+    to a DynamicMap.
+    """
+
+    callable_function = param.Callable(default=lambda x: x, doc="""
+         The callable function being wrapped.""")
+
+    inputs = param.List(default=[], doc="""
+         The list of inputs the callable function is wrapping.""")
+
+    def __call__(self, *args, **kwargs):
+        return self.callable_function(*args, **kwargs)
+
+
+def get_nested_streams(dmap):
+    """
+    Get all (potentially nested) streams from DynamicMap with Callable
+    callback.
+    """
+    layer_streams = list(dmap.streams)
+    if not isinstance(dmap.callback, Callable):
+        return layer_streams
+    for o in dmap.callback.inputs:
+        if isinstance(o, DynamicMap):
+            layer_streams += get_nested_streams(o)
+    return layer_streams
+
 
 class DynamicMap(HoloMap):
     """
@@ -689,7 +727,8 @@ def __getitem__(self, key):
 
         # Cache lookup
         try:
-            dimensionless = util.dimensionless_contents(self.streams, self.kdims)
+            dimensionless = util.dimensionless_contents(get_nested_streams(self),
+                                                        self.kdims, no_duplicates=False)
             if (dimensionless and not self._dimensionless_cache):
                 raise KeyError('Using dimensionless streams disables DynamicMap cache')
             cache = super(DynamicMap,self).__getitem__(key)

holoviews/core/util.py

@@ -799,21 +799,21 @@ def stream_parameters(streams, no_duplicates=True, exclude=['name']):
     return [name for name in names if name not in exclude]
 
 
-def dimensionless_contents(streams, kdims):
+def dimensionless_contents(streams, kdims, no_duplicates=True):
     """
     Return a list of stream parameters that have not been associated
     with any of the key dimensions.
     """
-    names = stream_parameters(streams)
+    names = stream_parameters(streams, no_duplicates)
     return [name for name in names if name not in kdims]
 
 
-def unbound_dimensions(streams, kdims):
+def unbound_dimensions(streams, kdims, no_duplicates=True):
     """
     Return a list of dimensions that have not been associated with
     any streams.
     """
-    params = stream_parameters(streams)
+    params = stream_parameters(streams, no_duplicates)
     return [d for d in kdims if d not in params]
 
 

holoviews/plotting/bokeh/element.py

@@ -31,7 +31,7 @@
 from ...element import RGB
 from ...streams import Stream, RangeXY, RangeX, RangeY
 from ..plot import GenericElementPlot, GenericOverlayPlot
-from ..util import dynamic_update
+from ..util import dynamic_update, get_sources
 from .plot import BokehPlot
 from .util import (mpl_to_bokeh, convert_datetime, update_plot,
                    bokeh_version, mplcmap_to_palette)
@@ -177,15 +177,18 @@ def _construct_callbacks(self):
         the plotted object as a source.
         """
         if not self.static or isinstance(self.hmap, DynamicMap):
-            source = self.hmap
+            sources = [(i, o) for i, o in get_sources(self.hmap)
+                       if i in [None, self.zorder]]
         else:
-            source = self.hmap.last
-        streams = Stream.registry.get(id(source), [])
-        registry = Stream._callbacks['bokeh']
-        callbacks = {(registry[type(stream)], stream) for stream in streams
-                     if type(stream) in registry and streams}
+            sources = [(self.zorder, self.hmap.last)]
+        cb_classes = set()
+        for _, source in sources:
+            streams = Stream.registry.get(id(source), [])
+            registry = Stream._callbacks['bokeh']
+            cb_classes |= {(registry[type(stream)], stream) for stream in streams
+                           if type(stream) in registry and streams}
         cbs = []
-        sorted_cbs = sorted(callbacks, key=lambda x: id(x[0]))
+        sorted_cbs = sorted(cb_classes, key=lambda x: id(x[0]))
         for cb, group in groupby(sorted_cbs, lambda x: x[0]):
             cb_streams = [s for _, s in group]
             cbs.append(cb(self, cb_streams, source))
@@ -560,6 +563,11 @@ def initialize_plot(self, ranges=None, plot=None, plots=None, source=None):
         if plot is None:
             plot = self._init_plot(key, style_element, ranges=ranges, plots=plots)
             self._init_axes(plot)
+        else:
+            self.handles['xaxis'] = plot.xaxis[0]
+            self.handles['x_range'] = plot.x_range
+            self.handles['y_axis'] = plot.yaxis[0]
+            self.handles['y_range'] = plot.y_range
         self.handles['plot'] = plot
 
         # Get data and initialize data source
@@ -675,7 +683,10 @@ def current_handles(self):
                 rangex, rangey = True, True
             elif isinstance(self.hmap, DynamicMap):
                 rangex, rangey = True, True
-                for stream in self.hmap.streams:
+                subplots = list(self.subplots.values()) if self.subplots else []
+                callbacks = [cb for p in [self]+subplots for cb in p.callbacks]
+                streams = [s for cb in callbacks for s in cb.streams]
+                for stream in streams:
                     if isinstance(stream, RangeXY):
                         rangex, rangey = False, False
                         break

holoviews/plotting/plot.py

@@ -22,7 +22,7 @@
 from ..core.util import stream_parameters
 from ..element import Table
 from .util import (get_dynamic_mode, initialize_sampled, dim_axis_label,
-                   attach_streams, traverse_setter)
+                   attach_streams, traverse_setter, get_nested_streams)
 
 
 class Plot(param.Parameterized):
@@ -578,7 +578,10 @@ def __init__(self, element, keys=None, ranges=None, dimensions=None,
                                                  **dict(params, **plot_opts))
         if top_level:
             self.comm = self.init_comm(element)
-        self.streams = self.hmap.streams if isinstance(self.hmap, DynamicMap) else []
+        streams = []
+        if isinstance(self.hmap, DynamicMap):
+            streams = get_nested_streams(self.hmap)
+        self.streams = streams
 
         # Update plot and style options for batched plots
         if self.batched:
@@ -928,9 +931,9 @@ def __init__(self, layout, keys=None, dimensions=None, **params):
         if top_level:
             self.comm = self.init_comm(layout)
         self.traverse(lambda x: setattr(x, 'comm', self.comm))
-        self.streams = [s for streams in layout.traverse(lambda x: x.streams,
-                                                         [DynamicMap])
-                        for s in streams]
+        nested_streams = layout.traverse(lambda x: get_nested_streams(x),
+                                         [DynamicMap])
+        self.streams = [s for streams in nested_streams for s in streams]
 
 
     def _get_frame(self, key):

holoviews/plotting/renderer.py

@@ -204,7 +204,7 @@ def _validate(self, obj, fmt):
             if (((len(plot) == 1 and not plot.dynamic)
                 or (len(plot) > 1 and self.holomap is None) or
                 (plot.dynamic and len(plot.keys[0]) == 0)) or
-                not unbound_dimensions(plot.streams, plot.dimensions)):
+                not unbound_dimensions(plot.streams, plot.dimensions, no_duplicates=False)):
                 fmt = fig_formats[0] if self.fig=='auto' else self.fig
             else:
                 fmt = holomap_formats[0] if self.holomap=='auto' else self.holomap

holoviews/plotting/util.py

@@ -4,7 +4,8 @@
 import param
 
 from ..core import (HoloMap, DynamicMap, CompositeOverlay, Layout,
-                    GridSpace, NdLayout, Store, Overlay)
+                    GridSpace, NdLayout, Store, Callable, Overlay)
+from ..core.spaces import get_nested_streams
 from ..core.util import (match_spec, is_number, wrap_tuple, basestring,
                          get_overlay_spec, unique_iterator, safe_unicode)
 
@@ -295,11 +296,36 @@ def attach_streams(plot, obj):
     Attaches plot refresh to all streams on the object.
     """
     def append_refresh(dmap):
-        for stream in dmap.streams:
+        for stream in get_nested_streams(dmap):
             stream._hidden_subscribers.append(plot.refresh)
     return obj.traverse(append_refresh, [DynamicMap])
 
 
+def get_sources(obj, index=None):
+    """
+    Traverses Callable graph to resolve sources on
+    DynamicMap objects, returning a list of sources
+    indexed by the Overlay layer.
+    """
+    layers = [(index, obj)]
+    if not isinstance(obj, DynamicMap) or not isinstance(obj.callback, Callable):
+        return layers
+    index = 0 if index is None else int(index)
+    for o in obj.callback.inputs:
+        if isinstance(o, Overlay):
+            layers.append((None, o))
+            for i, o in enumerate(overlay):
+                layers.append((index+i, o))
+            index += len(o)
+        elif isinstance(o, DynamicMap):
+            layers += get_sources(o, index)
+            index = layers[-1][0]+1
+        else:
+            layers.append((index, o))
+            index += 1
+    return layers
+
+
 def traverse_setter(obj, attribute, value):
     """
     Traverses the object and sets the supplied attribute on the

holoviews/plotting/widgets/__init__.py

@@ -108,7 +108,11 @@ def __init__(self, plot, renderer=None, **params):
         super(NdWidget, self).__init__(**params)
         self.id = plot.comm.target if plot.comm else uuid.uuid4().hex
         self.plot = plot
-        self.dimensions, self.keys = drop_streams(plot.streams,
+        streams = []
+        for stream in plot.streams:
+            if any(k in plot.dimensions for k in stream.contents):
+                streams.append(stream)
+        self.dimensions, self.keys = drop_streams(streams,
                                                   plot.dimensions,
                                                   plot.keys)
 

holoviews/util.py

@@ -5,6 +5,8 @@
 from .core import DynamicMap, ViewableElement
 from .core.operation import ElementOperation
 from .core.util import Aliases
+from .core.operation import OperationCallable
+from .core.spaces import Callable
 from .core import util
 from .streams import Stream
 
@@ -33,7 +35,8 @@ def __call__(self, map_obj, **params):
         self.p = param.ParamOverrides(self, params)
         callback = self._dynamic_operation(map_obj)
         if isinstance(map_obj, DynamicMap):
-            dmap = map_obj.clone(callback=callback, shared_data=False)
+            dmap = map_obj.clone(callback=callback, shared_data=False,
+                                 streams=[])
         else:
             dmap = self._make_dynamic(map_obj, callback)
         if isinstance(self.p.operation, ElementOperation):
@@ -44,7 +47,7 @@ def __call__(self, map_obj, **params):
                 elif not isinstance(stream, Stream):
                     raise ValueError('Stream must only contain Stream '
                                      'classes or instances')
-                stream.update(**{k: self.p.operation.p.get(k) for k, v in
+                stream.update(**{k: self.p.operation.p.get(k, v) for k, v in
                                  stream.contents.items()})
                 streams.append(stream)
             return dmap.clone(streams=streams)
@@ -69,15 +72,17 @@ def _dynamic_operation(self, map_obj):
             def dynamic_operation(*key, **kwargs):
                 self.p.kwargs.update(kwargs)
                 return self._process(map_obj[key], key)
-            return dynamic_operation
-
-        def dynamic_operation(*key, **kwargs):
-            key = key[0] if map_obj.mode == 'open' else key
-            self.p.kwargs.update(kwargs)
-            _, el = util.get_dynamic_item(map_obj, map_obj.kdims, key)
-            return self._process(el, key)
-
-        return dynamic_operation
+        else:
+            def dynamic_operation(*key, **kwargs):
+                key = key[0] if map_obj.mode == 'open' else key
+                self.p.kwargs.update(kwargs)
+                _, el = util.get_dynamic_item(map_obj, map_obj.kdims, key)
+                return self._process(el, key)
+        if isinstance(self.p.operation, ElementOperation):
+            return OperationCallable(callable_function=dynamic_operation,
+                                     inputs=[map_obj], operation=self.p.operation)
+        else:
+            return Callable(callable_function=dynamic_operation, inputs=[map_obj])
 
 
     def _make_dynamic(self, hmap, dynamic_fn):