My application folder structure is "data_visulization" which is base folder, inside data_visulization folder there are main.py and other folders are data and pages. data folder is having get_data.py and inside pages folder init.py and root.py and other folders are Analysis and Overview for the analysis.py and overview.py page. The main.py is only for all the pages calling. root.py is for displaying header of the page of KPI and data is fetch by call get_data.py for the KPI and For Analysis page and Overview page get_data.py is called separately. how to make this application to refresh the database data automatically and manually by providing 'Refresh Now' button on header page root.py. Below are the code which i tried, but unable to refresh the data.
get_data.py
# data/get_data.py
import psycopg2
import pandas as pd
from datetime import datetime, timedelta
DB_HOST = "localhost"
DB_PORT = "5432"
DB_NAME = "test"
DB_USER = "pg_db"
DB_PASS = "Admin"
data = None # Shared variable for all pages
filtered_df = None
def fetch_data():
conn = psycopg2.connect(
host=DB_HOST,
port=DB_PORT,
dbname=DB_NAME,
user=DB_USER,
password=DB_PASS
)
query = "SELECT * FROM prod_db"
data = pd.read_sql(query, conn)
data['date'] = pd.to_datetime(data['date'])
data.columns = ['Date', 'Location', 'Tower Name', 'Zone', 'Stage Label', 'Class Stage Label']
# Get yesterday's date
yesterday = (datetime.now() - timedelta(days=1)).date()
# Filter rows where date equals yesterday
filtered_df = data[data['Date'].dt.date == yesterday]
filtered_df = filtered_df.sort_values(by='Class Stage Label', ascending=False)
conn.close()
return data , filtered_df
def update_data(state=None):
global data , filtered_df
data , filtered_df = fetch_data()
if state: # For Taipy refresh calls
state.data = data
state.filtered_df = filtered_df
main.py
from taipy.gui import Gui
import threading
import time
from data.get_data import update_data
from pages import root
from Analysis import analysis
from Overview import overview
# Auto-refresh loop
def auto_refresh(state):
while True:
time.sleep(300) # every 5 minutes
update_data(state)
# Start auto-refresh when GUI initializes
def start_auto_refresh(state):
update_data(state) # load initial data
threading.Thread(target=auto_refresh, args=(state,), daemon=True).start()
pages = {
"/": root_page,
"Analysis": analysis_page,
"Overview": overview_page
}
Gui(pages=pages).run(on_init=start_auto_refresh)
root.py
from data.get_data import filtered_df, update_data
import taipy.gui.builder as tgb
with tgb.Page() as root_page:
tgb.text('## Performance Insights', mode='md')
tgb.button('Refresh Now', on_action = update_data)
with tgb.layout("1 1 1 1 1 1"):
with tgb.part():
tgb.text('### Date', mode='md')
tgb.text("#### {','.join(filtered_df['Date'].dt.strftime('%Y-%m-%d').unique())}", mode='md')
with tgb.part():
tgb.text('### Total', mode='md')
tgb.text("#### {int(filtered_df['Class Stage Label'].count())}", mode='md')
with tgb.part():
tgb.text('### Normal Stage', mode='md')
tgb.text("#### {int(filtered_df[filtered_df['Class Stage Label']=='Normal'].shape[0])}", mode='md')
with tgb.part():
tgb.text('### Stage One', mode='md')
tgb.text("#### {int(filtered_df[filtered_df['Class Stage Label']=='Stage1'].shape[0])}", mode='md')
with tgb.part():
tgb.text('### Stage Two', mode='md')
tgb.text("#### {int(filtered_df[filtered_df['Class Stage Label']=='Stage2'].shape[0])}", class_name="lblink", mode='md')
with tgb.part():
tgb.text('### Stage Three', mode='md')
tgb.text("#### {int(filtered_df[filtered_df['Class Stage Label']=='Stage3'].shape[0])}", class_name="blink", mode='md')
tgb.navbar()
with tgb.expandable(title ='Data', expanded=False):
tgb.table("{filtered_df}")
Try changing your main.py
to the following:
# main.py
from taipy.gui import Gui
import threading
import time
from data.get_data import update_data, data, filtered_df
from pages import root
from Analysis import analysis
from Overview import overview
# When new user session connects, update data
def on_init(state):
update_data(state)
# Server creates a separate thread to broadcast refreshed data to all users every 5 minutes
def auto_refresh(gui: Gui):
while True:
time.sleep(300) # every 5 minutes
gui.broadcast_callback(update_data)
pages = {
"/": root_page,
"Analysis": analysis_page,
"Overview": overview_page
}
gui = Gui(pages=pages)
threading.Thread(target=auto_refresh, args=(gui,)).start()
gui.run()
Note the changes:
Import data
and filtered_df
. By having them in the main module (where Gui is created), they are now in Taipy's "global" state, meaning that they are shared between all pages.
Using the on_init
callback to call update_data when a new session is initialized. Gui and <Gui>.run
do not have an on_init parameter. Taipy automatically looks for a function named on_init in the main module, but if you wanted to be explicit, you can do:
def on_init(state): ...
gui = Gui(page=page)
gui.on_init = on_init
gui.run()
Creating a single separate thread and using <Gui>.broadcast_callback
to update the data for all connected users every 5 minutes.