Skip to content
Advertisement

Is there a way to combine a loading data callback and a recurrent updating data callback with dash-python?

I am working on a dashboard with dash-plotly and I have created two different callbacks to handle data storage. The first callback loads initial data based on a dropdown

@app.callback(Output('battery-data','data'),
    [Input('battery-dropdown', 'value')],
    [State('battery-data','data')]
)
def load_data(battery_name):
    
    example_data = utils_new.load_dump_from_mongo(collection,battery_name)
    
    example_pd = pd.DataFrame(example_data)
    
    print(example_pd.head())

    example_pd = example_pd.drop('_id',axis=1)
    
    return(example_pd.to_json(date_format='iso', orient='split'))

While the second gets as input the initial data and then updates it recurrently

@app.callback(Output('battery-data', 'data'), 
            [Input('interval-component', 'n_intervals'),
             Input('battery-data','data'),
             Input('battery-dropdown', 'value']
)              [State('battery-data', 'data')])
def update_data(n, battery_json, battery_name, time_range):

    example_old_data = pd.read_json(battery_json, orient='split')

    example_old_data['Time'] = pd.to_datetime(example_old_data['Time']).apply(lambda x: x.replace(tzinfo=None))

    last_record_time = parser.parse(str(example_old_data['Time'].tail(1).item()))
    
    # load last 10 seconds of data as a list of records
    update_data = utils_new.update_data_from_mongo(collection,battery_name,last_record_time)
    
    update_pd = pd.DataFrame(update_data)

    print('SIZE OF THE UPDATE', len(update_pd))

    update_pd.drop("_id", axis=1, inplace=True)

    # add update data
    len_update = len(update_pd)
    
    example_new_data = example_old_data.append(update_pd,ignore_index=True)

    example_new_data.drop(example_new_data.head(len_update).index,inplace=True)

    example_new_data = example_new_data.reset_index(drop=True)

    # filter for battery id
    example_new_data_battery = example_new_data[example_new_data['Battery_ID'] == battery_name]

    return(example_new_data.to_json(date_format='iso', orient='split'))

app.layout = dbc.Container([
    dbc.Row([    
        html.H1(children='Battery monitoring'),
        html.Div(children='''
                This dashboard shows battery state 
            '''),
        html.Br(style={'marginBottom': '10px'})
            ]),
    dbc.Row([
        dbc.Col([
            dbc.Row([batteries_drop()]),
            ],width=6),
        dbc.Col([
            dbc.Row(id='card-anomaly'),
            ], width=6)]),
    dcc.Store(id='battery-data')
    dcc.Interval(id='interval-component',
                 interval=10*1000,
                 n_intervals=0)
])

However, in this way, it doesn’t work because I use the same ID for two different callbacks. Do you know if there is a way to make it work? Or if you know a strategy to address this? Initially, I would like to load only the data from a single battery to avoid crashing the dashboard and then I need to get the data updated every 10 secs.

Thanks for any suggestion or help!

Advertisement

Answer

You can fix this issue by using the great dash-extensions library.

This library has some modules that modify the Dash and allow us to extend it to do more than the standard Plotly does. By using a module called MultiplexerTransform combined with DashProxy you will be able to use the same ID in different callback outputs;

Below is a code snippet which does exactly what you did, you just need to adjust it to your context.

from dash.exceptions import PreventUpdate
from dash_extensions.enrich import Output, DashProxy, Input, MultiplexerTransform, html

app = DashProxy(transforms=[MultiplexerTransform()])
app.layout = html.Div([
    html.Button("Left", id="left"),
    html.Button("Right", id="right"),
    html.Div(id="log")
])

@app.callback(Output("log", "children"), Input("left", "n_clicks"))
def left(n_clicks):
    if not n_clicks:
        raise PreventUpdate()
    return "left"

@app.callback(Output("log", "children"), Input("right", "n_clicks"))
def right(n_clicks):
    if not n_clicks:
        raise PreventUpdate()
    return "right"

if __name__ == "__main__":
    app.run_server()

To read more about this module and library you can read: dash-extensions

Regards,
Leonardo

Advertisement