In [ ]:

    

import numpy as np
import time
import datetime

import collections

from bokeh.plotting import *
from bokeh.objects import Glyph


    

In [ ]:

    

output_notebook(url="default")


    

In [ ]:

    

TS_MULT_us = 1e6
UNIX_EPOCH = datetime.datetime(1970, 1, 1, 0, 0) #offset-naive datetime

def int2dt(ts, ts_mult=TS_MULT_us):
    """Convert timestamp (integer) to datetime"""
    return(datetime.datetime.utcfromtimestamp(float(ts)/ts_mult))
        
def td2int(td, ts_mult=TS_MULT_us):
    """Convert timedelta to integer"""
    return(int(td.total_seconds()*ts_mult))
        
def dt2int(dt, ts_mult=TS_MULT_us):
    """Convert datetime to integer"""
    delta = dt - UNIX_EPOCH
    return(int(delta.total_seconds()*ts_mult))
        
def int_from_last_sample(dt, td):
    return(dt2int(dt) - dt2int(dt) % td2int(td))


    

In [ ]:

    

TS_MULT = 1e3
td_delay = datetime.timedelta(seconds=0.5)
delay_s = td_delay.total_seconds()
delay_int = td2int(td_delay, TS_MULT)

value = 1000 # initial value
N = 100 # number of elements into circular buffer

buff = collections.deque([value]*N, maxlen=N)


    

In [ ]:

    

t_now = datetime.datetime.utcnow()
ts_now = dt2int(t_now, TS_MULT)
t = collections.deque(np.arange(ts_now-N*delay_int, ts_now, delay_int), maxlen=N)


    

In [ ]:

    

line(list(t), list(buff), color="#0000FF", tools="pan,wheel_zoom,box_zoom,reset,previewsave",
    name="line_example", x_axis_type="datetime"
)


    

In [ ]:

    

UNIX_EPOCH = datetime.datetime(1970, 1, 1, 0, 0) #offset-naive datetime


    

In [ ]:

    

renderer = [r for r in curplot().renderers if isinstance(r, Glyph)][0]
ds = renderer.data_source
show()
while(True):
    ts_now = dt2int(datetime.datetime.utcnow(), 1e3)
    t.append(ts_now)
    ds.data['x'] = list(t)

    value += np.random.uniform(-1, 1)
    buff.append(value)
    ds.data['y'] = list(buff)

    ds._dirty = True
    cursession().push(*ds.dump(docid=cursession().docid))
    
    time.sleep(delay_s)


    

In [ ]: