Skip to content

Instantly share code, notes, and snippets.

@Kruptein Kruptein/plot.py Secret

Created Mar 27, 2017
Embed
What would you like to do?
from datetime import datetime
from functools import partial
from multiprocessing import Process
from threading import Thread
from bokeh.application import Application
from bokeh.client import push_session
from bokeh.layouts import column
from bokeh.models import ColumnDataSource, Range1d, logging
from bokeh.plotting import figure, curdoc
from bokeh.server.server import Server
from tornado import gen
def bokeh_server():
server = Server({"/": Application()})
print("Start Server")
server.run_until_shutdown()
p = Process(target=bokeh_server)
p.start()
class StatefulPlot:
def __init__(self):
self.doc = curdoc()
self.session = push_session(self.doc)
self.plots = column()
self.doc.add_root(self.plots)
def add_plot(self, plot):
self.doc.add_next_tick_callback(partial(self._add_plot, plot))
@gen.coroutine
def _add_plot(self, plot):
self.plots.children.append(plot.fig)
def remove_plot(self, plot):
self.doc.add_next_tick_callback(partial(self._remove_plot, plot))
@gen.coroutine
def _remove_plot(self, plot):
self.plots.children.remove(plot.fig)
class Plot:
def __init__(self, title=''):
self.fig = figure(plot_width=800, plot_height=400, y_range=Range1d(-0.1, 1.1), title=title)
self.fig.xaxis.axis_label = "Time passed (s)"
self.fig.yaxis.axis_label = "Normalized value"
self.source = ColumnDataSource(data=dict(x=[], y1=[], y2=[], y3=[]))
self.linea = self.fig.line("x", "y1", source=self.source, legend="Potential", line_color="red")
self.lineb = self.fig.line("x", "y2", source=self.source, legend="Current")
self.linec = self.fig.line("x", "y3", source=self.source, legend="Power", line_color="green")
# self.queue = queue.Queue()
self.start_time = datetime.now()
fig.add_plot(self)
def update(self, y1, y2):
fig.doc.add_next_tick_callback(partial(self._update, y1, y2))
def remove(self):
fig.remove_plot(self)
@gen.coroutine
def _update(self, y1, y2):
timestamp = datetime.now()
new_x = ((timestamp - self.start_time).total_seconds())
data = dict(x=[new_x], y1=[y1], y2=[y2], y3=[y1 * y2])
self.source.stream(data, 75)
fig = StatefulPlot()
fig.session.show()
thread = Thread(target=fig.session.loop_until_closed)
thread.start()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.