# Path: antevents/base.py
# class Filter(Publisher, DefaultSubscriber):
#     def __init__(self, previous_in_chain,
#                  on_next=None, on_completed=None,
#                  on_error=None, name=None):
#         super().__init__()
#         self._on_next = on_next
#         self._on_completed = on_completed
#         self._on_error = on_error
#         self.name = name
#         self.dispose = previous_in_chain.subscribe(self) # XXX how to use this?
# 
#     def on_next(self, x):
#         if self._on_next:
#             try:
#                 self._on_next(self, x)
#             except FatalError:
#                 raise
#             except Exception as e:
#                 logger.exception("Got an exception on %s.on_next(%s)" %
#                                  (self, x))
#                 self.on_error(e)
#                 self.dispose() # stop from getting upstream events
#         else:
#             self._dispatch_next(x)
#         
#     def on_error(self, e):
#         if self._on_error:
#             self._on_error(self, e)
#         else:
#             self._dispatch_error(e)
#         
#     def on_completed(self):
#         if self._on_completed:
#             self._on_completed(self)
#         else:
#             self._dispatch_completed()
# 
#     def __str__(self):
#         if hasattr(self, 'name') and self.name:
#             return self.name
#         else:
#             return super().__str__()
# 
# def filtermethod(base, alias=None):
#     """Function decorator that creates a linq-style filter out of the
#     specified function. As described in the antevents.linq documentation,
#     it should take a Publisher as its first argument (the source of events)
#     and return a Publisher (representing the end the filter sequence once
#     the filter is included. The returned Publisher is typically an instance
#     of antevents.base.Filter.
# 
#     The specified function is used in two places:
#     1. A method with the specified name is added to the specified class
#        (usually the Publisher base class). This is for the fluent (method
#        chaining) API.
#     2. A function is created in the local namespace for use in the functional API.
#        This function does not take the publisher as an argument. Instead,
#        it takes the remaining arguments and then returns a function which,
#        when passed a publisher, subscribes to it and returns a filter.
# 
#     Decorator arguments:
#     :param T base: Base class to extend with method
#      (usually antevents.base.Publisher)
#     :param string alias: an alias for this function or list of aliases
#                          (e.g. map for select, etc.).
# 
#     :returns: A function that takes the class to be decorated.
#     :rtype: func -> func
# 
#     This was adapted from the RxPy extensionmethod decorator.
#     """
#     def inner(func):
#         """This function is returned by the outer filtermethod()
# 
#         :param types.FunctionType func: Function to be decorated
#         """
# 
#         func_names = [func.__name__,]
#         if alias:
#             aliases = alias if isinstance(alias, list) else [alias]
#             func_names += aliases
# 
#         _thunk = _ThunkBuilder(func)
# 
#         # For the primary name and all aliases, set the name on the
#         # base class as well as in the local namespace.
#         for func_name in func_names:
#             setattr(base, func_name, func)
#             func.__globals__[func_name] = _thunk
#         return _thunk
#     return inner


import datetime
import logging
import functools
import threading, queue
import numpy as np
from math import pi
from bokeh.charts import TimeSeries, show, output_file, output_server
from bokeh.plotting import figure, curdoc
from bokeh.layouts import column # to show two or more plots arranged in a column
from bokeh.models import ColumnDataSource
from bokeh.models import DatetimeTickFormatter
from bokeh.client import push_session
from antevents.base import Filter, filtermethod
                self.source.stream(new_data, 300)
                self.counter = 0
        except queue.Empty:
            pass
            self.counter = self.counter + 1
            if self.counter == 10:
                exit(0)

    def run(self):
        print("In thread.run")
        self.p = figure(plot_height=500, tools=TOOLS, y_axis_location='left', title=self.title)
        self.p.x_range.follow = "end"
        self.p.xaxis.axis_label = "Timestamp"
        self.p.x_range.follow_interval = 100
        self.p.x_range.range_padding = 0 
        self.p.line(x="timestamp", y="value", color="blue", source=self.source)
        self.p.circle(x="timestamp", y="value", color="red", source=self.source)

        self.session = push_session(curdoc())
        curdoc().add_periodic_callback(self.update, 100) #period in ms

        self.session.show(column(self.p)) 
        curdoc().title = 'Sensor' 
        self.session.loop_until_closed()

    # def register(self, d, sourceq):
    #     source = ColumnDataSource(dict(d))
    #     self.p.line(x=d[0], y=d[1], color="orange", source=source)
    #     curdoc().add_periodic_callback(self.update, 100) #period in ms


# Path: antevents/base.py
# class Filter(Publisher, DefaultSubscriber):
#     def __init__(self, previous_in_chain,
#                  on_next=None, on_completed=None,
#                  on_error=None, name=None):
#         super().__init__()
#         self._on_next = on_next
#         self._on_completed = on_completed
#         self._on_error = on_error
#         self.name = name
#         self.dispose = previous_in_chain.subscribe(self) # XXX how to use this?
# 
#     def on_next(self, x):
#         if self._on_next:
#             try:
#                 self._on_next(self, x)
#             except FatalError:
#                 raise
#             except Exception as e:
#                 logger.exception("Got an exception on %s.on_next(%s)" %
#                                  (self, x))
#                 self.on_error(e)
#                 self.dispose() # stop from getting upstream events
#         else:
#             self._dispatch_next(x)
#         
#     def on_error(self, e):
#         if self._on_error:
#             self._on_error(self, e)
#         else:
#             self._dispatch_error(e)
#         
#     def on_completed(self):
#         if self._on_completed:
#             self._on_completed(self)
#         else:
#             self._dispatch_completed()
# 
#     def __str__(self):
#         if hasattr(self, 'name') and self.name:
#             return self.name
#         else:
#             return super().__str__()
# 
# def filtermethod(base, alias=None):
#     """Function decorator that creates a linq-style filter out of the
#     specified function. As described in the antevents.linq documentation,
#     it should take a Publisher as its first argument (the source of events)
#     and return a Publisher (representing the end the filter sequence once
#     the filter is included. The returned Publisher is typically an instance
#     of antevents.base.Filter.
# 
#     The specified function is used in two places:
#     1. A method with the specified name is added to the specified class
#        (usually the Publisher base class). This is for the fluent (method
#        chaining) API.
#     2. A function is created in the local namespace for use in the functional API.
#        This function does not take the publisher as an argument. Instead,
#        it takes the remaining arguments and then returns a function which,
#        when passed a publisher, subscribes to it and returns a filter.
# 
#     Decorator arguments:
#     :param T base: Base class to extend with method
#      (usually antevents.base.Publisher)
#     :param string alias: an alias for this function or list of aliases
#                          (e.g. map for select, etc.).
# 
#     :returns: A function that takes the class to be decorated.
#     :rtype: func -> func
# 
#     This was adapted from the RxPy extensionmethod decorator.
#     """
#     def inner(func):
#         """This function is returned by the outer filtermethod()
# 
#         :param types.FunctionType func: Function to be decorated
#         """
# 
#         func_names = [func.__name__,]
#         if alias:
#             aliases = alias if isinstance(alias, list) else [alias]
#             func_names += aliases
# 
#         _thunk = _ThunkBuilder(func)
# 
#         # For the primary name and all aliases, set the name on the
#         # base class as well as in the local namespace.
#         for func_name in func_names:
#             setattr(base, func_name, func)
#             func.__globals__[func_name] = _thunk
#         return _thunk
#     return inner


# Path: antevents/adapters/bokeh.py
import datetime
import logging
import functools
import threading, queue
import numpy as np
from math import pi
from bokeh.charts import TimeSeries, show, output_file, output_server
from bokeh.plotting import figure, curdoc
from bokeh.layouts import column # to show two or more plots arranged in a column
from bokeh.models import ColumnDataSource
from bokeh.models import DatetimeTickFormatter
from bokeh.client import push_session
from antevents.base import Filter, filtermethod

                self.source.stream(new_data, 300)
                self.counter = 0
        except queue.Empty:
            pass
            self.counter = self.counter + 1
            if self.counter == 10:
                exit(0)

    def run(self):
        print("In thread.run")
        self.p = figure(plot_height=500, tools=TOOLS, y_axis_location='left', title=self.title)
        self.p.x_range.follow = "end"
        self.p.xaxis.axis_label = "Timestamp"
        self.p.x_range.follow_interval = 100
        self.p.x_range.range_padding = 0 
        self.p.line(x="timestamp", y="value", color="blue", source=self.source)
        self.p.circle(x="timestamp", y="value", color="red", source=self.source)

        self.session = push_session(curdoc())
        curdoc().add_periodic_callback(self.update, 100) #period in ms

        self.session.show(column(self.p)) 
        curdoc().title = 'Sensor' 
        self.session.loop_until_closed()

    # def register(self, d, sourceq):
    #     source = ColumnDataSource(dict(d))
    #     self.p.line(x=d[0], y=d[1], color="orange", source=source)
    #     curdoc().add_periodic_callback(self.update, 100) #period in ms


class BokehStreamer(Filter):