Merge pull request #404 from martindurant/small_bits

Pieces that occurred while working on intake-streamz

Author: martindurant

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,6 +228,16 @@ 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
+
+    def start(self):
+        self.stream.start()
+
+    def stop(self):
+        self.stream.stop()
+
     def _ipython_display_(self, **kwargs):
         return self.stream.latest().rate_limit(
             0.5).gather()._ipython_display_(**kwargs)

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:
@@ -548,13 +560,6 @@ def remove(self, predicate):
         """ Only pass through elements for which the predicate returns False """
         return self.filter(lambda x: not predicate(x))
 
-    def stop(self):
-        """Call on any stream node to halt all upstream sources"""
-        prev, s = self.upstream, self
-        while s:
-            prev, s = s, s.upstream
-        prev.stopped = True
-
     @property
     def scan(self):
         return self.accumulate

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
@@ -326,6 +326,10 @@ def concat(tup, module=None, columns=None):
                 example = kwargs.get('example')
             elif len(args) > 1:
                 example = args[1]
+            if callable(example):
+                example = example()
+                kwargs["example"] = example
+
             self._subtype = get_base_frame_type(self.__class__.__name__,
                                                 is_dataframe_like, example)
             super(DataFrame, self).__init__(*args, **kwargs)
@@ -337,6 +341,15 @@ def verify(self, x):
             raise IndexError("Input expected to have columns %s, got %s" %
                              (self.example.columns, x.columns))
 
+    @property
+    def plot(self):
+        try:
+            # import has side-effect of attaching .hvplot attribute
+            import hvplot.streamz  # # noqa: F401
+        except ImportError as err:  # pragma: no cover
+            raise ImportError("Streamz dataframe plotting requires hvplot") from err
+        return self.hvplot
+
 
 class _SeriesMixin(object):
     @property
@@ -801,8 +814,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 +853,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 +879,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.continue_ = [False]  # like the oppose of self.stopped
         self.kwargs = kwargs
 
         stream = self.source.map(lambda x: datafn(**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):
+        if not self.continue_[0]:
+            self.continue_[0] = True
+            self.loop.add_callback(self._cb, self.interval, self.source,
+                                   self.continue_)
 
     def __del__(self):
         self.stop()
@@ -891,16 +912,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
 
@@ -916,8 +937,9 @@ class Random(PeriodicDataFrame):
     """
 
     def __init__(self, freq='100ms', interval='500ms', dask=False,
-                 datafn=random_datablock):
-        super(Random, self).__init__(datafn, interval, dask, freq=pd.Timedelta(freq))
+                 start=True, datafn=random_datablock):
+        super(Random, self).__init__(datafn, interval, dask, start,
+                                     freq=pd.Timedelta(freq))
 
 
 _stream_types['streaming'].append((is_dataframe_like, DataFrame))

streamz/dataframe/tests/test_dataframes.py

@@ -557,7 +557,6 @@ def test_gc():
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
     n = len(sdf.stream.downstreams)
-    yield gen.sleep(0.1)
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
     yield gen.sleep(0.1)
@@ -566,6 +565,7 @@ def test_gc():
     yield gen.sleep(0.1)
     a = DataFrame({'volatility': sdf.x.rolling('100ms').var(),
                             'sub': sdf.x - sdf.x.rolling('100ms').mean()})
+    yield gen.sleep(0.1)
 
     assert len(sdf.stream.downstreams) == n
     del a

streamz/tests/test_sources.py

@@ -141,10 +141,10 @@ def test_from_iterable():
 def test_from_iterable_backpressure():
     it = iter(range(5))
     source = Source.from_iterable(it)
-    L = source.rate_limit(0.01).sink_to_list()
+    L = source.rate_limit(0.1).sink_to_list()
     source.start()
 
-    wait_for(lambda: L == [0], 1)
+    wait_for(lambda: L == [0], 1, period=0.01)
     assert next(it) == 2  # 1 is in blocked _emit