Hello everyone,
I have developed a Dash Plotly app that visualizes data from my database. Now, I want to add a scheduler app to my existing Dash app that can run certain background tasks periodically, such as fetching data, sending email alerts, or performing calculations.
I have found some example code for the scheduler app and I understand how it works, but I’m not sure where exactly to put it in my Dash app and how to run it effectively.
Here is the code for the scheduler app:
import time
import schedule
import win32com
database = constants.database
# Function to send email
def send_email(subject, message, to):
outlook = win32com.client.Dispatch('Outlook.Application')
mail = outlook.CreateItem(0)
mail.Subject = subject
mail.To = to
mail.Body = message
mail.Send()
# Function to fetch the downtime
def fetch_the_downtime():
start_time = datetime.datetime.now() - datetime.timedelta(days=1)
end_time = datetime.datetime.now()
parameters = [start_time, end_time]
for db in database:
query_string = mssql_queries.build_db_query_app7()
records = mssql_conn.execute_query(query_string, parameters, str(db))
if records:
dataframe = pd.DataFrame.from_records(records)
dataframe.columns = ["ID", "Timestamp", "Duration", "Week Number", "FailureCode", "FailureDescription"]
dataframe["Duration"] = (
dataframe["Duration"].astype("datetime64[s]").dt.strftime("%H:%M:%S")
)
# Check if any downtime duration is over 3600 seconds
if (dataframe['Duration'] > '01:00:00').any():
send_email('Downtime Alert', f"{db} - Some downtime duration is over 3600 seconds.", 'some mail')
# Refresh the function every 30 minutes
while True:
fetch_the_downtime()
time.sleep(1800) # 1800 seconds = 30 minutes
#Function to check scrap rate
def check_scrap_rate():
# Connect to the databases
conn_dict = {}
for db in database:
while True:
# Calculate the time window for scrap rate calculation
now = datetime.datetime.now()
one_hour_ago = now - datetime.timedelta(hours=1)
# Check the scrap rate for each database
for db in database:
# Execute the database query to get the scrap rate
query = f'''
SELECT COUNT(*) AS total_items, SUM(CASE WHEN [Status] = 3 THEN 1 ELSE 0 END) AS scrap_items
FROM [{db}].[dbo].[ItemLog]
WHERE [Type] = 1 AND [Timestamp] BETWEEN '{one_hour_ago}' AND '{now}'
'''
cursor = conn_dict[db].cursor()
cursor.execute(query)
result = cursor.fetchone()
# Calculate the scrap rate
if result:
total_items = result[0]
scrap_items = result[1]
if total_items > 0:
scrap_rate = 100 * scrap_items / total_items
else:
scrap_rate = 0
else:
scrap_rate = 0
# Check if the scrap rate of the last hour is higher than 3% or the scrap rate
# in the last 500 pieces produced is higher than 3%
if scrap_rate > 3:
subject = f"{db} more than 3% scrap"
message = f"Scrap rate of last hour is {scrap_rate:.2f}%, which is higher than 3%."
send_email(subject, message, 'some mail')
else:
# Execute the database query to get the number of items produced in the last 500 pieces
query = f'''
SELECT COUNT(*) AS produced_items, SUM(CASE WHEN [Status] = 3 THEN 1 ELSE 0 END) AS scrap_items
FROM [{db}].[dbo].[ItemLog]
WHERE [Type] = 1 ORDER BY [Timestamp] DESC LIMIT 500
'''
cursor = conn_dict[db].cursor()
cursor.execute(query)
result = cursor.fetchone()
# Calculate the scrap rate in the last 500 pieces
if result:
produced_items = result[0]
scrap_items = result[1]
if produced_items > 0:
scrap_rate = 100 * scrap_items / produced_items
else:
scrap_rate = 0
else:
scrap_rate = 0
# Check if the scrap rate in the last 500 pieces is higher than 3%
if scrap_rate > 3:
subject = f"{db} more than 3% scrap in last 500 pieces"
message = f"Scrap rate in last 500 pieces is {scrap_rate:.2f}%, which is higher than 3%."
send_email(subject, message, 'some mail')
# Wait for 30 minutes before checking
time.sleep(1800)
# Schedule the job to run every day at 8 AM
schedule.every().day.at("08:00").do(check_scrap_rate)
# Loop to run the scheduled jobs
# while True:
# schedule.run_pending()
# time.sleep(1)
In my Dash app, I have an index.py file where I define the layout and callbacks for my app. Should I add the scheduler code in this file? Or should I create a separate file for the scheduler and somehow run it alongside my Dash app?
index.py
from dash import dcc
from dash import html
from dash.dependencies import Input, Output
from app import app
from apps import app1, app2, app3, app4, app5, app6, app7, app8, scheduler
import nav
app.layout = html.Div(
[dcc.Location(id="url", refresh=False), html.Div(id="page-content")]
)
# "complete" layout
app.validation_layout = html.Div(
[
app1.layout,
app2.layout,
app3.layout,
app4.layout,
app5.layout,
app6.layout,
app7.layout,
app8.layout,
nav.layout,
app.layout,
]
)
@app.callback(Output("page-content", "children"), [Input("url", "pathname")])
def display_page(pathname):
if pathname == "/":
return nav.layout
elif pathname == "/apps/Downtime_and_Reasons":
# app.title = "Downtime and Reasons"
return app1.layout
elif pathname == "/apps/Production_history":
# app.title = "Report of Check Criteria"
return app2.layout
elif pathname == "/apps/Quantity_and_Scrap":
# app.title = "Quantity and Scrap"
return app3.layout
elif pathname == "/apps/User_Log":
# app.title = "User Log"
return app4.layout
elif pathname == "/apps/Error_Log":
# app.title = "Error Log"
return app5.layout
elif pathname == "/apps/Error_Log":
# app.title = "Error Log"
return app5.layout
elif pathname == "/apps/Parameter_Log":
# app.title = "Parameter Log"
return app6.layout
elif pathname == "/apps/Downtime_Log":
# app.title = "Downtime Log"
return app7.layout
elif pathname == "/apps/Overall_Equipment_Effectiveness":
# app.title = "Overall Equipment Effectiveness"
return app8.layout
else:
return "404"
if __name__ == "__main__":
app.run_server(debug=True, port=8080)
I would greatly appreciate it if someone could guide me on where exactly to place this scheduler code within my Dash app structure and how to run it effectively. I want to ensure that my Dash app runs smoothly and the scheduler app performs its tasks in the background without impacting the user experience.
Thank you in advance for your help!