I would like to pause my live trend when i zoom in. now, when i zoom in, it keeps the scale as zoomed in, but the trend is moving and i can not monitor the point i would want to observe.
import dash
from dash.dependencies import Output, Input, State
import dash_core_components as dcc
import dash_html_components as html
import plotly
import plotly.graph_objs as go
import random
from collections import deque
#import opcuaclient
from datetime import datetime
from datetime import timedelta
#import plotly.express as px
from flask import Flask
‘’’
Program workflow
changed dropdown -> change plot_list names and load data to plot_data-> redraw graph
(live mode) 1s callback reads data from OPC updates plot_data -> redraw graph
(historian mode) load maximum available data to plot_data -> redraw graph
‘’’
class ObjectHoldingTheValue:
def init(self, initial_value=0):
self._value = initial_value
self._callbacks = []
@property
def value(self):
return self._value
@value.setter
def value(self, new_value):
old_value = self._value
self._value = new_value
self._notify_observers(old_value, new_value)
def _notify_observers(self, old_value, new_value):
for callback in self._callbacks:
callback(old_value, new_value)
def register_callback(self, callback):
self._callbacks.append(callback)
def my_value_change_callback(old_value, new_value): # custom callback for value change -------------------------
print(‘callback activated’)
def timecheck():
now = datetime.now()
timecheck.current_time = now.strftime("%H:%M:%S")
before_now = now - timedelta(0, 60)
timecheck.before_time = before_now.strftime("%H:%M:%S")
future = now + timedelta(0, 60)
timecheck.future_time = future.strftime("%H:%M:%S")
def plot_data_limit():
while len(plot_data[‘Data0’][‘time’]) > max_small_data: plot_data[‘Data0’][‘time’].pop(0)
while len(plot_data[‘Data0’][‘value’]) > max_small_data: plot_data[‘Data0’][‘value’].pop(0)
while len(plot_data[‘Data1’][‘time’]) > max_small_data: plot_data[‘Data1’][‘time’].pop(0)
while len(plot_data[‘Data1’][‘value’]) > max_small_data: plot_data[‘Data1’][‘value’].pop(0)
while len(plot_data[‘Data2’][‘time’]) > max_small_data: plot_data[‘Data2’][‘time’].pop(0)
while len(plot_data[‘Data2’][‘value’]) > max_small_data: plot_data[‘Data2’][‘value’].pop(0)
while len(plot_data[‘Data3’][‘time’]) > max_small_data: plot_data[‘Data3’][‘time’].pop(0)
while len(plot_data[‘Data3’][‘value’]) > max_small_data: plot_data[‘Data3’][‘value’].pop(0)
def sim_add_DB():
timecheck()
data_DB [‘time’].append(timecheck.current_time)
data_DB[‘value’].append(random.uniform(0,10))
data_DB1[‘time’].append(timecheck.current_time)
data_DB1[‘value’].append(random.uniform(0, 10))
def sim_add_live():
timecheck()
plot_data[‘Data0’][‘time’].append(timecheck.current_time)
plot_data[‘Data0’][‘value’].append(random.uniform(0,10))
plot_data[‘Data1’][‘time’].append(timecheck.current_time)
plot_data[‘Data1’][‘value’].append(random.uniform(0, 10))
plot_data[‘Data2’][‘time’].append(timecheck.current_time)
plot_data[‘Data2’][‘value’].append(random.uniform(0, 10))
plot_data[‘Data3’][‘time’].append(timecheck.current_time)
plot_data[‘Data3’][‘value’].append(random.uniform(0, 10))
plot_data_limit()
def sim_init():
OPC_name_list.append(‘Temperature’)
OPC_name_list.append(‘Pressure’)
i = 0
now = datetime.now() - timedelta(0, max_small_data)
while i < max_small_data:
before_now = now + timedelta(0, i)
data_DB[‘time’][i] = before_now.strftime("%H:%M:%S")
data_DB[‘value’][i] = i
data_DB1[‘time’][i] = before_now.strftime("%H:%M:%S")
data_DB1[‘value’][i] = (i+1)/5
i = i + 1
def sim_data_base():
i=0
len_names = len(plot_data[‘names’])
while i < len_names:
data_name = 'Data' + str(i)
if plot_data['names'][i] == "Temperature":
lenght_data = len(data_DB['time'])
fromX = lenght_data - max_small_data
j = 0
while j < max_small_data:
plot_data[data_name]['time'][j] = data_DB['time'][j + fromX]
plot_data[data_name]['value'][j] = data_DB['value'][j + fromX]
j = j + 1
elif plot_data['names'][i]== "Pressure":
lenght_data = len(data_DB1['time'])
fromX = lenght_data - max_small_data
j = 0
while j < max_small_data:
plot_data[data_name]['time'][j] = data_DB1['time'][j + fromX]
plot_data[data_name]['value'][j] = data_DB1['value'][j + fromX]
j = j + 1
i=i+1
my_value_with_callback = ObjectHoldingTheValue() #custom callback with value cange ----------------------------
my_value_with_callback.register_callback(my_value_change_callback)
max_small_data=10
#cl = opcuaclient.opcconnect(address=“opc.tcp://127.0.0.1:9999/freeopcua/server/”)
timecheck()
data_DB = {
‘time’: [timecheck.before_time] * max_small_data, # because of time scale for first inicialisation
‘value’: [1] * max_small_data,
}
data_DB1 = {
‘time’: [timecheck.before_time] * max_small_data, # because of time scale for first inicialisation
‘value’: [1] * max_small_data,
}
plot_data = {
‘names’ : [],
‘Data0’ : {‘time’ : [0]*max_small_data,
‘value’: [0]*max_small_data},
‘Data1’: {‘time’: [0]*max_small_data,
‘value’: [0]*max_small_data},
‘Data2’: {‘time’: [0]*max_small_data,
‘value’: [0]*max_small_data},
‘Data3’: {‘time’: [0]*max_small_data,
‘value’: [0]*max_small_data},
‘Live_view’ : True,
‘Auto_scale’: False
}
plot_data[‘Data0’][‘time’] = plot_data[‘Data0’][‘time’]
plot_data[‘Data0’][‘value’] = plot_data[‘Data0’][‘value’]
plot_data[‘Data1’][‘time’] = plot_data[‘Data1’][‘time’]
plot_data[‘Data1’][‘value’] = plot_data[‘Data1’][‘value’]
plot_data[‘Data2’][‘time’] = plot_data[‘Data2’][‘time’]
plot_data[‘Data2’][‘value’] = plot_data[‘Data2’][‘value’]
plot_data[‘Data3’][‘time’] = plot_data[‘Data3’][‘time’]
plot_data[‘Data3’][‘value’] = plot_data[‘Data3’][‘value’]
OPC_name_list=[]
sim_init()
server = Flask(name)
external_stylesheets = [‘https://codepen.io/chriddyp/pen/bWLwgP.css’]
app = dash.Dash(name, external_stylesheets=external_stylesheets, server = server)
app.config.suppress_callback_exceptions = True
app.scripts.config.serve_locally = True
app.css.config.serve_locally = True
Drop_down_option = []
i=0
while i<len(OPC_name_list):
Drop_down_option.append({‘label’ : OPC_name_list[i], ‘value’ : OPC_name_list[i]})
i=i+1
app.layout = html.Div([
dcc.Graph(id=‘live-graph’, animate=False, config={‘scrollzoom’ : True
},
figure=go.Figure(
data=[
go.Scatter(
#x=data_DB[‘time’],
#y=data_DB[‘value’],
)
],
layout=go.Layout(
xaxis={#‘range’: [‘13:00:00’, ‘14:00:00’],
#‘autorange’: plot_data[‘Auto_scale’],
‘tickmode’: ‘auto’,
‘nticks’: 4,
},
yaxis={#‘range’: [0 , 1],
#‘autorange’: plot_data[‘Auto_scale’],
‘tickmode’: ‘auto’,
‘nticks’: 3,
},
uirevision= ‘refresh’
)
)
),
dcc.Interval(
id='graph-update',
interval=2 * 1000,
n_intervals=0
),
html.Label('Dropdown'),
dcc.Dropdown(id='Dropdown-list',
options=list(Drop_down_option),
value='',
multi=True
),
html.Button('Refresh', id='refresh', n_clicks=1),
#html.Button('Pause', id='pause', n_clicks=1),
#html.pre(id='output')
], style={‘columnCount’: 1})
@app.callback(dash.dependencies.Output(‘refresh’, ‘n_clicks’), # update data from OPC to Plot data
[dash.dependencies.Input(‘Dropdown-list’, ‘value’),
dash.dependencies.Input(‘graph-update’, ‘n_intervals’)
],
[dash.dependencies.State(‘refresh’, ‘n_clicks’)],
)
def data_update(names,n,state):
sim_add_DB()
if plot_data[‘names’]==names:
if plot_data[‘live_view’]== True:
print(‘add data’)
sim_add_live() # change to function that would add data from OPC --------------------------------------
return 1
else:
if len(names) > 0:
print(‘new data’)
plot_data[‘names’] = names
if plot_data[‘Live_view’]:
a=1
# later add function that would add data from data base that other tags have --------------------------
return 1
else:
sim_data_base() # change to function that would read data from data base ------------------------------------------------
return 1
else:
plot_data[‘names’] = []
a=2
return state
-------------------------------------------------------------------------------------------------------
@app.callback( #redraw graph
Output(‘live-graph’, ‘figure’),
[Input(‘refresh’, ‘n_clicks’)], # [Input(‘pause’, ‘n_clicks’)], for pause button
[State(‘live-graph’, ‘figure’)],
#prev_inputs=[
# prevInput(‘Refresh’, ‘n_clicks’),
# prevInput(‘Pause’, ‘n_clicks’)
#]
)
def update_graph(n_clicks,fig):
figure = go.Figure(
data=[
go.Scatter(
x=[0],
y=[0]
)
], )
Layout_plot = go.Layout(
xaxis={#‘autorange’: plot_data[‘Auto_scale’],
‘tickmode’: ‘auto’,
‘nticks’: 9,
},
yaxis={#‘autorange’: plot_data[‘Auto_scale’],
‘tickmode’: ‘auto’,
‘nticks’: 9,
},
uirevision= ‘refresh’
)
if n_clicks == 1:
plot_data['live_view']= True
print('refreshed')
print(n_clicks)
i = 0
if len(plot_data['names']) > 0:
while i < len(plot_data['names']):
data_name = 'Data' + str(i)
if (i == 0):
figure = go.Figure(
data=[
go.Scatter(
x=list(plot_data[data_name]['time']),
y=list(plot_data[data_name]['value']),
name = plot_data['names'][i]
)
],)
elif plot_data['names'][i] != []:
figure.add_trace(go.Scatter(
x=list(plot_data[data_name]['time']),
y=list(plot_data[data_name]['value']),
name=plot_data['names'][i]
))
i = i + 1
a=1
elif n_clicks == 2 :
plot_data['live_view']=False
i = 0
if len(plot_data['names']) > 0:
while i < len(plot_data['names']):
data_name = 'Data' + str(i)
if (i == 0):
figure = go.Figure(
data=[
go.Scatter(
x=list(plot_data[data_name]['time']-i),
y=list(plot_data[data_name]['value']-i),
name = plot_data['names'][i]
)
],)
elif plot_data['names'][i] != []:
figure.add_trace(go.Scatter(
x=list(plot_data[data_name]['time'-i]),
y=list(plot_data[data_name]['value'-i]),
name=plot_data['names'][i]
))
i = i + 1
b=1
print(n_clicks)
return figure
return {
'data' : figure['data'],
'layout' :Layout_plot#['uirevision']
}
if name == ‘main’:
#app.run_server(debug=True)
server.run()