current_value plus DataFrame methods

Author: Martin Durant

streamz/collection.py

@@ -152,7 +152,7 @@ def __rxor__(self, other):
         return self.map_partitions(operator.xor, other, self)
 
 
-class Streaming(OperatorMixin):
+class Streaming(OperatorMixin, core.APIRegisterMixin):
     """
     Superclass for streaming collections
 
@@ -228,12 +228,22 @@ def _repr_html_(self):
 
         return "<h5>%s - elements like<h5>\n%s" % (type(self).__name__, body)
 
+    @property
+    def current_value(self):
+        return self.stream.current_value
+
+    @current_value.setter
+    def current_value(self, *args, **kwargs):
+        # this actually happens in self.stream - Streaming.emit is ignored
+        pass
+
     def _ipython_display_(self, **kwargs):
         return self.stream.latest().rate_limit(
             0.5).gather()._ipython_display_(**kwargs)
 
     def emit(self, x):
         self.verify(x)
+        self.current_value = x
         self.stream.emit(x)
 
     def verify(self, x):

streamz/core.py

@@ -122,7 +122,79 @@ def __str__(self):
     __repr__ = __str__
 
 
-class Stream(object):
+class APIRegisterMixin(object):
+
+    @classmethod
+    def register_api(cls, modifier=identity, attribute_name=None):
+        """ Add callable to Stream API
+
+        This allows you to register a new method onto this class.  You can use
+        it as a decorator.::
+
+            >>> @Stream.register_api()
+            ... class foo(Stream):
+            ...     ...
+
+            >>> Stream().foo(...)  # this works now
+
+        It attaches the callable as a normal attribute to the class object.  In
+        doing so it respects inheritance (all subclasses of Stream will also
+        get the foo attribute).
+
+        By default callables are assumed to be instance methods.  If you like
+        you can include modifiers to apply before attaching to the class as in
+        the following case where we construct a ``staticmethod``.
+
+            >>> @Stream.register_api(staticmethod)
+            ... class foo(Stream):
+            ...     ...
+
+            >>> Stream.foo(...)  # Foo operates as a static method
+
+        You can also provide an optional ``attribute_name`` argument to control
+        the name of the attribute your callable will be attached as.
+
+            >>> @Stream.register_api(attribute_name="bar")
+            ... class foo(Stream):
+            ...     ...
+
+            >> Stream().bar(...)  # foo was actually attached as bar
+        """
+        def _(func):
+            @functools.wraps(func)
+            def wrapped(*args, **kwargs):
+                return func(*args, **kwargs)
+            name = attribute_name if attribute_name else func.__name__
+            setattr(cls, name, modifier(wrapped))
+            return func
+        return _
+
+    @classmethod
+    def register_plugin_entry_point(cls, entry_point, modifier=identity):
+        if hasattr(cls, entry_point.name):
+            raise ValueError(
+                f"Can't add {entry_point.name} from {entry_point.module_name} "
+                f"to {cls.__name__}: duplicate method name."
+            )
+
+        def stub(*args, **kwargs):
+            """ Entrypoints-based streamz plugin. Will be loaded on first call. """
+            node = entry_point.load()
+            if not issubclass(node, Stream):
+                raise TypeError(
+                    f"Error loading {entry_point.name} "
+                    f"from module {entry_point.module_name}: "
+                    f"{node.__class__.__name__} must be a subclass of Stream"
+                )
+            if getattr(cls, entry_point.name).__name__ == "stub":
+                cls.register_api(
+                    modifier=modifier, attribute_name=entry_point.name
+                )(node)
+            return node(*args, **kwargs)
+        cls.register_api(modifier=modifier, attribute_name=entry_point.name)(stub)
+
+
+class Stream(APIRegisterMixin):
     """ A Stream is an infinite sequence of data.
 
     Streams subscribe to each other passing and transforming data between them.
@@ -179,6 +251,8 @@ def __init__(self, upstream=None, upstreams=None, stream_name=None,
                  loop=None, asynchronous=None, ensure_io_loop=False):
         self.name = stream_name
         self.downstreams = OrderedWeakrefSet()
+        self.current_value = None
+        self.current_metadata = None
         if upstreams is not None:
             self.upstreams = list(upstreams)
         elif upstream is not None:
@@ -267,80 +341,16 @@ def _remove_upstream(self, upstream):
         classes which handle stream specific buffers/caches"""
         self.upstreams.remove(upstream)
 
-    @classmethod
-    def register_api(cls, modifier=identity, attribute_name=None):
-        """ Add callable to Stream API
-
-        This allows you to register a new method onto this class.  You can use
-        it as a decorator.::
-
-            >>> @Stream.register_api()
-            ... class foo(Stream):
-            ...     ...
-
-            >>> Stream().foo(...)  # this works now
-
-        It attaches the callable as a normal attribute to the class object.  In
-        doing so it respects inheritance (all subclasses of Stream will also
-        get the foo attribute).
-
-        By default callables are assumed to be instance methods.  If you like
-        you can include modifiers to apply before attaching to the class as in
-        the following case where we construct a ``staticmethod``.
-
-            >>> @Stream.register_api(staticmethod)
-            ... class foo(Stream):
-            ...     ...
-
-            >>> Stream.foo(...)  # Foo operates as a static method
-
-        You can also provide an optional ``attribute_name`` argument to control
-        the name of the attribute your callable will be attached as.
-
-            >>> @Stream.register_api(attribute_name="bar")
-            ... class foo(Stream):
-            ...     ...
-
-            >> Stream().bar(...)  # foo was actually attached as bar
-        """
-        def _(func):
-            @functools.wraps(func)
-            def wrapped(*args, **kwargs):
-                return func(*args, **kwargs)
-            name = attribute_name if attribute_name else func.__name__
-            setattr(cls, name, modifier(wrapped))
-            return func
-        return _
-
-    @classmethod
-    def register_plugin_entry_point(cls, entry_point, modifier=identity):
-        if hasattr(cls, entry_point.name):
-            raise ValueError(
-                f"Can't add {entry_point.name} from {entry_point.module_name} "
-                f"to {cls.__name__}: duplicate method name."
-            )
-
-        def stub(*args, **kwargs):
-            """ Entrypoints-based streamz plugin. Will be loaded on first call. """
-            node = entry_point.load()
-            if not issubclass(node, Stream):
-                raise TypeError(
-                    f"Error loading {entry_point.name} "
-                    f"from module {entry_point.module_name}: "
-                    f"{node.__class__.__name__} must be a subclass of Stream"
-                )
-            if getattr(cls, entry_point.name).__name__ == "stub":
-                cls.register_api(
-                    modifier=modifier, attribute_name=entry_point.name
-                )(node)
-            return node(*args, **kwargs)
-        cls.register_api(modifier=modifier, attribute_name=entry_point.name)(stub)
-
     def start(self):
         """ Start any upstream sources """
         for upstream in self.upstreams:
             upstream.start()
 
+    def stop(self):
+        """ Stop upstream sources """
+        for upstream in self.upstreams:
+            upstream.stop()
+
     def __str__(self):
         s_list = []
         if self.name:
@@ -430,6 +440,8 @@ def _emit(self, x, metadata=None):
             A reference counter used to check when data is done
 
         """
+        self.current_value = x
+        self.current_metadata = metadata
         if metadata:
             self._retain_refs(metadata, len(self.downstreams))
         else:

streamz/dataframe/core.py

@@ -1,11 +1,11 @@
-from __future__ import division, print_function
+import asyncio
 
 import operator
 from collections import OrderedDict
 import numpy as np
 import pandas as pd
 import toolz
-from tornado.ioloop import IOLoop
+
 from tornado import gen
 
 from ..collection import Streaming, _stream_types, OperatorMixin
@@ -801,8 +801,10 @@ def _accumulate(self, Agg, **kwargs):
         return Streaming(outstream, example, stream_type=stream_type)
 
 
-def random_datapoint(now, **kwargs):
+def random_datapoint(now=None, **kwargs):
     """Example of querying a single current value"""
+    if now is None:
+        now = pd.Timestamp.now()
     return pd.DataFrame(
         {'a': np.random.random(1)}, index=[now])
 
@@ -838,6 +840,7 @@ def random_datablock(last, now, **kwargs):
     return df
 
 
+@DataFrame.register_api(staticmethod, "from_periodic")
 class PeriodicDataFrame(DataFrame):
     """A streaming dataframe using the asyncio ioloop to poll a callback fn
 
@@ -863,26 +866,31 @@ class PeriodicDataFrame(DataFrame):
     >>> df = PeriodicDataFrame(interval='1s', datafn=random_datapoint)  # doctest: +SKIP
     """
 
-    def __init__(self, datafn=random_datablock, interval='500ms', dask=False, **kwargs):
+    def __init__(self, datafn=random_datablock, interval='500ms', dask=False,
+                 start=True, **kwargs):
         if dask:
             from streamz.dask import DaskStream
             source = DaskStream()
-            loop = source.loop
         else:
             source = Source()
-            loop = IOLoop.current()
+        self.loop = source.loop
         self.interval = pd.Timedelta(interval).total_seconds()
         self.source = source
         self.continue_ = [True]
         self.kwargs = kwargs
 
-        stream = self.source.map(lambda x: datafn(**x, **kwargs))
+        stream = self.source.map(
+            lambda x, continue_=self.continue_: datafn(continue_=continue_, **x, **kwargs)
+        )
         example = datafn(last=pd.Timestamp.now(), now=pd.Timestamp.now(), **kwargs)
 
         super(PeriodicDataFrame, self).__init__(stream, example)
+        if start:
+            self.start()
 
-        loop.add_callback(self._cb, self.interval, self.source,
-                          self.continue_)
+    def start(self):
+        self.loop.add_callback(self._cb, self.interval, self.source,
+                               self.continue_)
 
     def __del__(self):
         self.stop()
@@ -891,16 +899,16 @@ def stop(self):
         self.continue_[0] = False
 
     @staticmethod
-    @gen.coroutine
-    def _cb(interval, source, continue_):
+    async def _cb(interval, source, continue_):
         last = pd.Timestamp.now()
         while continue_[0]:
-            yield gen.sleep(interval)
+            await gen.sleep(interval)
             now = pd.Timestamp.now()
-            yield source._emit(dict(last=last, now=now))
+            await asyncio.gather(*source._emit(dict(last=last, now=now)))
             last = now
 
 
+@DataFrame.register_api(staticmethod, "random")
 class Random(PeriodicDataFrame):
     """PeriodicDataFrame providing random values by default