Non-blocking periodic_events method on DynamicMap

holoviews/core/spaces.py

@@ -17,6 +17,8 @@
 from .options import Store, StoreOptions
 from ..streams import Stream
 
+
+
 class HoloMap(UniformNdMapping, Overlayable):
     """
     A HoloMap can hold any number of DataLayers indexed by a list of
@@ -557,18 +559,23 @@ def __call__(self):
             raise
 
 
+def get_nested_dmaps(dmap):
+    """
+    Get all DynamicMaps referenced by the supplied DynamicMap's callback.
+    """
+    dmaps = [dmap]
+    for o in dmap.callback.inputs:
+        if isinstance(o, DynamicMap):
+            dmaps.extend(get_nested_dmaps(o))
+    return list(set(dmaps))
+
+
 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 list(set(layer_streams))
-    for o in dmap.callback.inputs:
-        if isinstance(o, DynamicMap):
-            layer_streams += get_nested_streams(o)
-    return list(set(layer_streams))
+    return list({s for dmap in get_nested_dmaps(dmap) for s in dmap.streams})
 
 
 @contextmanager
@@ -589,6 +596,53 @@ def dynamicmap_memoization(callable_obj, streams):
         callable_obj.memoize = memoization_state
 
 
+
+class periodic(object):
+    """
+    Implements the utility of the same name on DynamicMap.
+
+    Used to defined periodic event updates that can be started and
+    stopped.
+    """
+    _periodic_util = util.periodic
+
+    def __init__(self, dmap):
+        self.dmap = dmap
+        self.instance = None
+
+    def __call__(self, period, count, param_fn=None):
+        """
+        Run a non-blocking loop that updates the stream parameters using
+        the event method. Runs count times with the specified period. If
+        count is None, runs indefinitely.
+
+        If param_fn is not specified, the event method is called without
+        arguments. If it is specified, it must be a callable accepting a
+        single argument (the iteration count) that returns a dictionary
+        of the new stream values to be passed to the event method.
+        """
+
+        if self.instance is not None and not self.instance.completed:
+            raise RuntimeError('Periodic process already running. '
+                               'Wait until it completes or call '
+                               'stop() before running a new periodic process')
+        def inner(i):
+            kwargs = {} if param_fn is None else param_fn(i)
+            self.dmap.event(**kwargs)
+
+        instance = self._periodic_util(period, count, inner)
+        instance.start()
+        self.instance= instance
+
+    def stop(self):
+        "Stop the periodic process."
+        self.instance.stop()
+
+    def __str__(self):
+        return "<holoviews.core.spaces.periodic method>"
+
+
+
 class DynamicMap(HoloMap):
     """
     A DynamicMap is a type of HoloMap where the elements are dynamically
@@ -661,6 +715,7 @@ def __init__(self, callback, initial_items=None, **params):
             if stream.source is None:
                 stream.source = self
         self.redim = redim(self, mode='dynamic')
+        self.periodic = periodic(self)
 
     @property
     def unbounded(self):
@@ -754,7 +809,6 @@ def event(self, **kwargs):
 
         Stream.trigger(self.streams)
 
-
     def _style(self, retval):
         """
         Use any applicable OptionTree of the DynamicMap to apply options

holoviews/core/util.py

@@ -10,6 +10,7 @@
 from functools import partial
 from contextlib import contextmanager
 
+from threading import Thread, Event
 import numpy as np
 import param
 
@@ -78,6 +79,49 @@ def default(self, obj):
             return id(obj)
 
 
+class periodic(Thread):
+    """
+    Run a callback count times with a given period without blocking.
+    """
+
+    def __init__(self, period, count, callback):
+
+        if isinstance(count, int):
+            if count < 0: raise ValueError('Count value must be positive')
+        elif not type(count) is type(None):
+            raise ValueError('Count value must be a positive integer or None')
+
+        super(periodic, self).__init__()
+        self.period = period
+        self.callback = callback
+        self.count = count
+        self.counter = 0
+        self._completed = Event()
+
+    @property
+    def completed(self):
+        return self._completed.is_set()
+
+    def stop(self):
+        self._completed.set()
+
+    def __repr__(self):
+        return 'periodic(%s, %s, %s)' % (self.period,
+                                         self.count,
+                                         callable_name(self.callback))
+    def __str__(self):
+        return repr(self)
+
+    def run(self):
+        while not self.completed:
+            self._completed.wait(self.period)
+            self.callback(self.counter)
+            self.counter += 1
+
+            if self.counter == self.count:
+                self._completed.set()
+
+
 
 def deephash(obj):
     """

holoviews/plotting/bokeh/renderer.py

@@ -16,9 +16,10 @@
 
 from ...core import Store, HoloMap
 from ..comms import JupyterComm, Comm
+from ..plot import GenericElementPlot
 from ..renderer import Renderer, MIME_TYPES
 from .widgets import BokehScrubberWidget, BokehSelectionWidget, BokehServerWidgets
-from .util import compute_static_patch, serialize_json
+from .util import compute_static_patch, serialize_json, attach_periodic
 
 
 
@@ -122,9 +123,11 @@ def server_doc(self, plot, doc=None):
         if doc is None:
             doc = curdoc()
         if isinstance(plot, BokehServerWidgets):
-            plot.plot.document = doc
-        else:
-            plot.document = doc
+            plot = plot.plot
+        plot.document = doc
+        plot.traverse(lambda x: attach_periodic(plot),
+                      [GenericElementPlot])
+
         doc.add_root(plot.state)
         return doc
 

holoviews/plotting/bokeh/util.py

@@ -27,6 +27,7 @@
 from ...core.options import abbreviated_exception
 from ...core.overlay import Overlay
 from ...core.util import basestring, unique_array
+from ...core.spaces import get_nested_dmaps, DynamicMap
 
 from ..util import dim_axis_label, rgb2hex
 
@@ -664,3 +665,72 @@ def categorize_array(array, dim):
     treats as a categorical suffix.
     """
     return np.array([dim.pprint_value(x).replace(':', ';') for x in array])
+
+
+class periodic(object):
+    """
+    Mocks the API of periodic Thread in hv.core.util, allowing a smooth
+    API transition on bokeh server.
+    """
+
+    def __init__(self, document):
+        self.document = document
+        self.callback = None
+        self.period = None
+        self.count = None
+        self.counter = None
+
+    @property
+    def completed(self):
+        return self.counter is None
+
+    def start(self):
+        if self.document is None:
+            raise RuntimeError('periodic was registered to be run on bokeh'
+                               'server but no document was found.')
+        self.document.add_periodic_callback(self._periodic_callback, self.period)
+
+    def __call__(self, period, count, callback):
+        if isinstance(count, int):
+            if count < 0: raise ValueError('Count value must be positive')
+        elif not type(count) is type(None):
+            raise ValueError('Count value must be a positive integer or None')
+
+        self.callback = callback
+        self.period = period*1000.
+        self.count = count
+        self.counter = 0
+        return self
+
+    def _periodic_callback(self):
+        self.callback(self.counter)
+        self.counter += 1
+
+        if self.counter == self.count:
+            self.stop()
+
+    def stop(self):
+        self.counter = None
+        try:
+            self.document.remove_periodic_callback(self._periodic_callback)
+        except ValueError: # Already stopped
+            pass
+
+    def __repr__(self):
+        return 'periodic(%s, %s, %s)' % (self.period,
+                                         self.count,
+                                         callable_name(self.callback))
+    def __str__(self):
+        return repr(self)
+
+
+
+
+def attach_periodic(plot):
+    """
+    Attaches plot refresh to all streams on the object.
+    """
+    def append_refresh(dmap):
+        for dmap in get_nested_dmaps(dmap):
+            dmap.periodic._periodic_util = periodic(plot.document)
+    return plot.hmap.traverse(append_refresh, [DynamicMap])