How to save and get state in callback ()?

Question

I am trying to simulate read data from a queue, such as kafka. I need to keep a pointer to the current record when I transfer data from a file. I am currently doing this with a global variable, but it seems to be used for all user sessions.

How can I save this user state in bokeh?

def modify_doc(doc): df_all = pd.read_csv('data.csv') df_all['Date'] = pd.to_datetime(df_all['Date']) start_data = df_all[0:10].to_dict(orient='list') source = ColumnDataSource(data=start_data) ... def callback(): # FIXME: how can we save the current_record in the user session? global current_record try: current_record except NameError: current_record = 10 df = df_all[current_record:current_record+1] if df.shape[0] > 0: # we have another record so display it new_data = df.to_dict(orient='list') source.stream( new_data ) current_record = current_record + 1 doc.add_root(plot) doc.add_periodic_callback(callback, 250) 

I saw the documentation for ClientSession , but does this seem to work at all document level?

---

I have included a minimal, complete and verifiable example below:

file: bokeh_server.py

Run locally with: python3 bokeh_server.py

 import pandas as pd from tornado.ioloop import IOLoop import yaml from jinja2 import Template from bokeh.application.handlers import FunctionHandler from bokeh.application import Application from bokeh.layouts import column from bokeh.models import ColumnDataSource, Slider, Div from bokeh.plotting import figure from bokeh.server.server import Server from bokeh.themes import Theme from bokeh.client import push_session import os # if running locally, listen on port 5000 PORT = int(os.getenv('PORT', '5000')) HOST = "0.0.0.0" try: # This is set in the cloud foundry manifest. If we are running on # cloud foundry, this will be set for us. ALLOW_WEBSOCKET_ORIGIN = os.getenv("ALLOW_WEBSOCKET_ORIGIN").split(',') except: # We are not running on cloud foundry so we must be running locally ALLOW_WEBSOCKET_ORIGIN = [ 'localhost:{0}'.format(PORT) ] io_loop = IOLoop.current() # This example simulates reading from a stream such as kafka def modify_doc(doc): df_all = pd.read_csv('data.csv') df_all['Date'] = pd.to_datetime(df_all['Date']) start_data = df_all[0:10].to_dict(orient='list') source = ColumnDataSource(data=start_data) plot = figure(x_axis_type='datetime', y_range=(0, 10000000), y_axis_label='Y Label', title="Title") plot.line('Date', 'ALL_EXCL_FUEL', color='blue', alpha=1, source=source) plot.line('Date', 'MOSTLY_FOOD', color='lightblue', alpha=1, source=source) plot.line('Date', 'NON_SPECIALISED', color='grey', alpha=1, source=source) def callback(): # FIXME: how can we save this in the user session? global counter try: counter except NameError: counter = 10 df = df_all[counter:counter+1] if df.shape[0] > 0: # hardcode update values for now new_data = df.to_dict(orient='list') source.stream( new_data ) counter = counter + 1 doc.add_root(plot) doc.add_periodic_callback(callback, 250) bokeh_app = Application(FunctionHandler(modify_doc)) server = Server( {'/': bokeh_app}, io_loop=io_loop, allow_websocket_origin=ALLOW_WEBSOCKET_ORIGIN, **{'port': PORT, 'address': HOST} ) server.start() if __name__ == '__main__': io_loop.add_callback(server.show, "/") io_loop.start() 

file: data.csv

 Date,ALL_EXCL_FUEL,MOSTLY_FOOD,NON_SPECIALISED,TEXTILE,HOUSEHOLD,OTHER,NON_STORE 1986 Jan,1883154,747432,163708,267774,261453,281699,161088 1986 Feb,1819796,773161,152656,223836,246502,275121,148519 1986 Mar,1912582,797104,169440,251438,249614,292348,152638 1986 Apr,1974419,809334,170540,275975,260086,299271,159213 1986 May,1948915,800193,170173,274979,251175,297655,154740 1986 Jun,2019114,821785,178366,295463,251507,311447,160546 1986 Jul,2051539,816033,184812,297969,269786,323187,159752 1986 Aug,2011746,804386,180911,297138,263427,310220,155665 1986 Sep,2046678,792943,181055,305350,280640,318368,168322 1986 Oct,2110669,810147,187728,308919,298637,325617,179621 1986 Nov,2315710,847794,231599,352009,332079,358077,194152 1986 Dec,2830206,970987,319570,490001,373714,469399,206536 1987 Jan,2032021,798562,172215,288186,288534,307900,176624 1987 Feb,1980748,805713,165682,247219,282836,313577,165721 1987 Mar,2009717,816051,174034,256756,280207,315562,167106 1987 Apr,2156967,862749,189729,308543,284440,336755,174751 1987 May,2075808,834375,175464,287515,280404,330093,167957 1987 Jun,2137092,844051,183014,304706,286522,345149,173651 1987 Jul,2208377,847098,198848,330804,301537,356037,174054 

Answer 1

I did some testing and found that every time a new browser session was opened using the bokeh chart diagram, a new instance of the bokeh Document was created. The answer for me was to keep state in the instance of the document:

 def modify_doc(doc): # The first 100 records of data.csv will be loaded immediately # The remaining records will be read one-by-one in the update # callback which is used to simulate new, realtime data arriving doc.realtime_rec_ptr = 100 df_all = pd.read_csv('data.csv') df_all['Date'] = pd.to_datetime(df_all['Date']) start_data_df = df_all[0:doc.realtime_rec_ptr] start_data_df.loc[ :, 'color' ] = 'green' src = ColumnDataSource(data=start_data_df.to_dict(orient='list')) p = figure(x_axis_type='datetime', title="Title" y_range=(0, 10000000), y_axis_label='Y Label') p.line('Date','ALL_EXCL_FUEL',color='blue',alpha=1,source=src) # realtime markers will be colored green, others will be blue p.circle('Date','ALL_EXCL_FUEL',color='color',fill_alpha=0.2,size=4,source=src) def callback(): df = df_all[doc.realtime_rec_ptr:realtime_rec_ptr+1] if df.shape[0] > 0: df.loc[ :, 'color' ] = 'blue' new_data = df.to_dict(orient='list') #print(new_data) source.stream( new_data ) doc.realtime_rec_ptr = doc.realtime_rec_ptr + 1 doc.add_root(p) doc.add_periodic_callback(callback, 250)