Replaced Stream.cleanup with resetting machinery

holoviews/streams.py

@@ -32,20 +32,53 @@ def triggering_streams(streams):
             stream._triggering = False
 
 
+@contextmanager
+def disable_constant(parameterized):
+    """
+    Temporarily set parameters on Parameterized object to
+    constant=False.
+    """
+    params = parameterized.params().values()
+    constants = [p.constant for p in params]
+    for param in params:
+        param.constant = False
+    try:
+        yield
+    except:
+        raise
+    finally:
+        for (param, const) in zip(params, constants):
+            param.constant = const
+
 
 class Stream(param.Parameterized):
     """
     A Stream is simply a parameterized object with parameters that
-    change over time in response to update events. Parameters are
-    updated via the update method.
-
-    Streams may have one or more subscribers which are callables passed
-    the parameter dictionary when the trigger classmethod is called.
-
-    Depending on the plotting backend certain streams may interactively
-    subscribe to events and changes by the plotting backend. For this
-    purpose use the LinkedStream baseclass, which enables the linked
-    option by default.
+    change over time in response to update events and may trigger
+    downstream events on its subscribers. The subscribers may be
+    supplied as a list of callables or added later add_subscriber
+    method. The subscriber will be called whenever the Stream is
+    triggered and passed a dictionary mapping of the parameters of the
+    stream, which are available on the instance as the ``contents``.
+
+    Depending on the plotting backend certain streams may
+    interactively subscribe to events and changes by the plotting
+    backend. For this purpose use the LinkedStream baseclass, which
+    enables the linked option by default. A source for the linking may
+    be supplied to the constructor in the form of another viewable
+    object specifying which part of a plot the data should come from.
+
+    Since a Stream may represent a transient event you may disable
+    stream parameters from being memoized, indicating that each stream
+    event should trigger an update in a downstream DynamicMap callback.
+    By enabling resets on the constructor the reset method will get
+    called each time the stream is triggered, resetting all Stream
+    parameters to their defaults.
+
+    The Stream class is meant for subclassing and should at minimum
+    declare a number of parameters but may also override the transform
+    and reset method to preprocess parameters before they are passed
+    to subscribers and reset them using custom logic respectively.
     """
 
     # Mapping from a source id to a list of streams
@@ -83,19 +116,13 @@ def trigger(cls, streams):
                 subscriber(**dict(union))
 
         for stream in streams:
-            params = stream.params().values()
-            constants = [p.constant for p in params]
-            for param in params:
-                param.constant = False
-
-            stream.deactivate()
-
-            for (param, const) in zip(params, constants):
-                param.constant = const
+            with disable_constant(stream):
+                if stream._reset:
+                    stream.reset()
 
 
     def __init__(self, rename={}, source=None, subscribers=[], linked=False,
-                 memoize=True, **params):
+                 memoize=True, reset=False, **params):
         """
         The rename argument allows multiple streams with similar event
         state to be used by remapping parameter names.
@@ -115,6 +142,7 @@ def __init__(self, rename={}, source=None, subscribers=[], linked=False,
         self.memoize = memoize
         self.linked = linked
         self._rename = self._validate_rename(rename)
+        self._reset = reset
 
         # Whether this stream is currently triggering its subscribers
         self._triggering = False
@@ -134,12 +162,24 @@ def subscribers(self):
         " Property returning the subscriber list"
         return self._subscribers
 
+
     def clear(self):
         """
         Clear all subscribers registered to this stream.
         """
         self._subscribers = []
 
+
+    def reset(self):
+        """
+        Resets stream parameters to their defaults.
+        """
+        with disable_constant(self):
+            for k, p in self.params().items():
+                if k != 'name':
+                    setattr(self, k, p.default)
+
+
     def add_subscriber(self, subscriber):
         """
         Register a callable subscriber to this stream which will be
@@ -150,6 +190,7 @@ def add_subscriber(self, subscriber):
             raise TypeError('Subscriber must be a callable.')
         self._subscribers.append(subscriber)
 
+
     def _validate_rename(self, mapping):
         param_names = [k for k in self.params().keys() if k != 'name']
         for k,v in mapping.items():
@@ -160,6 +201,7 @@ def _validate_rename(self, mapping):
                                'stream parameter of the same name' % v)
         return mapping
 
+
     def rename(self, **mapping):
         """
         The rename method allows stream parameters to be allocated to
@@ -172,19 +214,11 @@ def rename(self, **mapping):
                               source=self._source,
                               linked=self.linked, **params)
 
-
-    def deactivate(self):
-        """
-        Allows defining an action after the stream has been triggered,
-        e.g. resetting parameters on streams with transient events.
-        """
-        pass
-
-
     @property
     def source(self):
         return self._source
 
+
     @source.setter
     def source(self, source):
         if self._source:
@@ -203,6 +237,7 @@ def transform(self):
         """
         return {}
 
+
     @property
     def contents(self):
         filtered = {k:v for k,v in self.get_param_values() if k!= 'name' }
@@ -214,19 +249,8 @@ def _set_stream_parameters(self, **kwargs):
         Sets the stream parameters which are expected to be declared
         constant.
         """
-        params = self.params().values()
-        constants = [p.constant for p in params]
-        for param in params:
-            param.constant = False
-        try:
+        with disable_constant(self) as constant:
             self.set_param(**kwargs)
-        except Exception as e:
-            for (param, const) in zip(params, constants):
-                param.constant = const
-            raise
-
-        for (param, const) in zip(params, constants):
-            param.constant = const
 
 
     def update(self, trigger=True, **kwargs):
@@ -246,6 +270,7 @@ def update(self, trigger=True, **kwargs):
         if trigger:
             self.trigger([self])
 
+
     def __repr__(self):
         cls_name = self.__class__.__name__
         kwargs = ','.join('%s=%r' % (k,v)
@@ -280,8 +305,9 @@ class PositionX(LinkedStream):
     position of the mouse/trackpad cursor.
     """
 
-    x = param.ClassSelector(class_=(Number, util.basestring), default=0, doc="""
-           Position along the x-axis in data coordinates""", constant=True)
+    x = param.ClassSelector(class_=(Number, util.basestring), default=None,
+                            constant=True, doc="""
+           Position along the x-axis in data coordinates""")
 
 
 class PositionY(LinkedStream):
@@ -292,8 +318,9 @@ class PositionY(LinkedStream):
     position of the mouse/trackpad cursor.
     """
 
-    y = param.ClassSelector(class_=(Number, util.basestring), default=0, doc="""
-           Position along the y-axis in data coordinates""", constant=True)
+    y = param.ClassSelector(class_=(Number, util.basestring), default=None,
+                            constant=True, doc="""
+           Position along the y-axis in data coordinates""")
 
 
 class PositionXY(LinkedStream):
@@ -304,11 +331,13 @@ class PositionXY(LinkedStream):
     position of the mouse/trackpad cursor.
     """
 
-    x = param.ClassSelector(class_=(Number, util.basestring), default=0, doc="""
-           Position along the x-axis in data coordinates""", constant=True)
+    x = param.ClassSelector(class_=(Number, util.basestring), default=None,
+                            constant=True, doc="""
+           Position along the x-axis in data coordinates""")
 
-    y = param.ClassSelector(class_=(Number, util.basestring), default=0, doc="""
-           Position along the y-axis in data coordinates""", constant=True)
+    y = param.ClassSelector(class_=(Number, util.basestring), default=None,
+                            constant=True, doc="""
+           Position along the y-axis in data coordinates""")
 
 
 class Tap(PositionXY):

tests/testdynamic.py

@@ -4,7 +4,7 @@
 from holoviews import Dimension, NdLayout, GridSpace
 from holoviews.core.spaces import DynamicMap, HoloMap, Callable
 from holoviews.element import Image, Scatter, Curve, Text
-from holoviews.streams import PositionXY, PositionX
+from holoviews.streams import PositionXY, PositionX, PositionY
 from holoviews.util import Dynamic
 from holoviews.element.comparison import ComparisonTestCase
 
@@ -206,7 +206,7 @@ def fn(x, y):
             dmap.event(x=1, y=2)
 
 
-class DynamicCallable(ComparisonTestCase):
+class DynamicCallableMemoize(ComparisonTestCase):
 
     def test_dynamic_callable_memoize(self):
         # Always memoized only one of each held
@@ -278,6 +278,42 @@ def history_callback(x, history=deque(maxlen=10)):
         self.assertEqual(dmap[()], Curve([1, 1, 1, 2, 2, 2]))
 
 
+
+class DynamicStreamReset(ComparisonTestCase):
+
+    def test_dynamic_stream_reset(self):
+        # Ensure Stream reset option resets streams to default value
+        # when not triggering
+        global xresets, yresets
+        xresets, yresets = 0, 0
+        def history_callback(x, y, history=deque(maxlen=10)):
+            global xresets, yresets
+            if x is None:
+                xresets += 1
+            else:
+                history.append(x)
+            if y is None:
+                yresets += 1
+
+            return Curve(list(history))
+
+        x = PositionX(reset=True, memoize=False, x=None)
+        y = PositionY(reset=True, memoize=False, y=None)
+        dmap = DynamicMap(history_callback, kdims=[], streams=[x, y])
+
+        # Add stream subscriber mocking plot
+        x.add_subscriber(lambda **kwargs: dmap[()])
+        y.add_subscriber(lambda **kwargs: dmap[()])
+
+        # Update each stream and count when None default appears
+        for i in range(2):
+            x.update(x=i)
+            y.update(y=i)
+
+        self.assertEqual(xresets, 2)
+        self.assertEqual(yresets, 2)
+
+
 class DynamicCollate(ComparisonTestCase):
 
     def test_dynamic_collate_layout(self):