tch / app.py
prudhviLatha's picture
Update app.py
ed0f40f verified
import os
import logging
import gradio as gr
import pandas as pd
import numpy as np
from datetime import datetime
import plotly.express as px
import plotly.graph_objects as go
import io
import base64
import matplotlib.pyplot as plt
from matplotlib.backends.backend_pdf import PdfPages
from simple_salesforce import Salesforce
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('labour_forecast.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Configuration class
class Config:
REQUIRED_COLUMNS = ['Date', 'Attendance', 'Trade', 'Weather']
ENCODINGS = ['utf-8', 'utf-8-sig', 'latin1', 'iso-8859-1', 'cp1252', 'utf-16']
FORECAST_DAYS = 3
WMA_WEIGHTS = {3: np.array([0.5, 0.3, 0.2]), 2: np.array([0.6, 0.4]), 1: np.array([1.0])}
WEATHER_IMPACT = {'Sunny': 0.0, 'Rainy': 0.25, 'Cloudy': 0.15, 'N/A': 0.05}
WEEKEND_ADJUSTMENT = 0.8
VALID_ALERT_STATUSES = {'Normal', 'Critical', 'Warning'}
# Salesforce connection
def connect_to_salesforce():
try:
sf = Salesforce(
username=os.getenv('SF_USERNAME'),
password=os.getenv('SF_PASSWORD'),
security_token=os.getenv('SF_SECURITY_TOKEN'),
domain=os.getenv('SF_DOMAIN', 'login')
)
logger.info("Successfully connected to Salesforce")
return sf
except Exception as e:
logger.error(f"Failed to connect to Salesforce: {str(e)}")
return None
# Data processing
def process_csv(file_path):
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error(error_msg)
return None, error_msg
for encoding in Config.ENCODINGS:
try:
df = pd.read_csv(file_path, encoding=encoding, dtype_backend='numpy_nullable')
df.columns = df.columns.str.strip().str.capitalize()
missing_columns = [col for col in Config.REQUIRED_COLUMNS if col not in df.columns]
if missing_columns:
raise ValueError(f"Missing columns: {', '.join(missing_columns)}")
df['Date'] = pd.to_datetime(df['Date'], dayfirst=True, errors='coerce')
if df['Date'].isna().all():
raise ValueError("All dates in CSV are invalid")
df['Attendance'] = pd.to_numeric(df['Attendance'], errors='coerce').fillna(0).astype('Int64')
for col in ['Weather', 'Trade']:
df[col] = df[col].astype(str).replace('nan', 'N/A')
logger.info(f"Successfully processed CSV with encoding '{encoding}'. Rows: {len(df)}. Attendance summary: {df['Attendance'].describe().to_dict()}")
return df, None
except Exception as e:
logger.warning(f"Failed with encoding '{encoding}': {str(e)}")
continue
error_msg = "Could not decode CSV file. Tried encodings: " + ", ".join(Config.ENCODINGS) + ". Ensure the file is a valid CSV and uses a supported encoding."
logger.error(error_msg)
return None, error_msg
# Forecasting logic with real-time adjustments
def weighted_moving_average_forecast(df, trade, site_calendar_date):
try:
site_calendar_date = pd.to_datetime(site_calendar_date)
trade_df = df[df['Trade'].str.lower() == trade.lower()].copy()
if trade_df.empty:
return [], [], None, 'N/A', 'Normal', f"No data found for trade: {trade}"
is_weekday = site_calendar_date.weekday() < 5
site_calendar = 1 if is_weekday else 0
# Filter data up to the selected date
trade_df = trade_df[trade_df['Date'] <= site_calendar_date]
recent_data = trade_df.tail(30)[['Date', 'Attendance', 'Weather']]
if recent_data.empty:
return [], [], None, 'N/A', 'Normal', f"No data for trade {trade} on or before {site_calendar_date.strftime('%Y-%m-%d')}"
# Real-time attendance for the selected date
current_day_data = trade_df[trade_df['Date'] == site_calendar_date]
if current_day_data.empty:
return [], [], None, 'N/A', 'Normal', f"No data for trade {trade} on {site_calendar_date.strftime('%Y-%m-%d')}"
current_attendance = current_day_data['Attendance'].iloc[0]
current_weather = current_day_data['Weather'].iloc[0]
predictions = []
shortage_probs = []
suggested_actions = 'Monitor'
alert_status = 'Normal'
future_dates = pd.date_range(site_calendar_date, periods=Config.FORECAST_DAYS + 1, freq='D')[1:]
# Calculate historical metrics
attendance_mean = recent_data['Attendance'].mean()
attendance_trend = recent_data['Attendance'].pct_change().mean() if len(recent_data) > 1 else 0
attendance_volatility = recent_data['Attendance'].pct_change().std() if len(recent_data) > 1 else 0.1
logger.info(f"Trade: {trade}, Mean Attendance: {attendance_mean}, Trend: {attendance_trend}, Volatility: {attendance_volatility}")
scale_factor = 1.0 if attendance_mean == 0 else min(100 / attendance_mean, 2.0)
# Calculate shortage risk for the current day
weather_impact = Config.WEATHER_IMPACT.get(current_weather, 0.05)
expected_attendance = attendance_mean * (1 - weather_impact) * (1 if site_calendar == 1 else Config.WEEKEND_ADJUSTMENT)
shortage_ratio = 1 - (current_attendance / expected_attendance) if expected_attendance > 0 else 0
shortage_prob = 0.5 + (shortage_ratio * 0.5) + (weather_impact * 0.3) + (attendance_trend * 0.2 * scale_factor)
shortage_prob = min(max(shortage_prob, 0.4), 0.9)
shortage_probs.append(shortage_prob)
if shortage_prob > 0.7:
suggested_actions = 'Urgent hiring needed'
alert_status = 'Critical'
elif shortage_prob > 0.5:
suggested_actions = 'Reschedule tasks'
alert_status = 'Warning'
else:
suggested_actions = 'Monitor'
alert_status = 'Normal'
# Forecast for future days
for i, date in enumerate(future_dates):
recent_attendance = recent_data['Attendance'].values[-3:]
weights = Config.WMA_WEIGHTS.get(len(recent_attendance), Config.WMA_WEIGHTS[1])
forecast_value = np.average(recent_attendance, weights=weights)
weather_idx = i % len(recent_data['Weather'])
future_weather = Config.WEATHER_IMPACT.get(recent_data['Weather'].iloc[-weather_idx-1], 0.05)
forecast_value *= (1 - future_weather)
headcount = round(forecast_value * (1 if site_calendar == 1 else Config.WEEKEND_ADJUSTMENT))
base_prob = 0.5 + (attendance_trend * 0.5 * scale_factor)
day_adjustment = (i + 1) * 0.02 * attendance_volatility
weather_adjustment = future_weather * 0.3
future_shortage_prob = base_prob + day_adjustment + weather_adjustment
future_shortage_prob = min(max(future_shortage_prob * 0.7 + 0.3, 0.5), 0.7)
shortage_probs.append(future_shortage_prob)
predictions.append({"date": date.strftime('%Y-%m-%d'), "headcount": headcount})
logger.info(f"Trade: {trade}, Shortage Probabilities: {shortage_probs}")
site_calendar_value = site_calendar_date.strftime('%Y-%m-%d') + f" ({'Weekday' if is_weekday else 'Weekend'})"
logger.info(f"Forecast generated for trade: {trade}")
return predictions, shortage_probs, site_calendar_value, suggested_actions, alert_status, None
except Exception as e:
logger.error(f"Forecast error for trade {trade}: {str(e)}")
return [], [], None, 'N/A', 'Normal', f"Forecast error: {str(e)}"
# Real-time shortage risk heatmap for the selected day
def create_heatmap(df, predictions_dict, shortage_probs_dict, site_calendar_date):
try:
site_calendar_date = pd.to_datetime(site_calendar_date)
heatmap_data = []
# Extend to 6 days to match the screenshot (2025-04-24 to 2025-04-29)
future_dates = pd.date_range(site_calendar_date, periods=6, freq='D')
for trade in predictions_dict.keys():
probs = shortage_probs_dict.get(trade, [0.5] * len(future_dates))
for i, date in enumerate(future_dates):
# Use the shortage probability for the current day (index 0) and future days
prob = probs[i] if i < len(probs) else probs[-1] # Fallback to last prob if not enough data
heatmap_data.append({
'Date': date.strftime('%Y-%m-%d'),
'Trade': trade,
'Shortage_Probability': prob
})
heatmap_df = pd.DataFrame(heatmap_data)
if heatmap_df.empty:
return go.Figure().update_layout(title="Shortage Risk Heatmap (No Data)")
display_probs = heatmap_df['Shortage_Probability'] * 100
# Custom colorscale adjusted to make 46% and 50% red
custom_colorscale = [
[0, 'red'], # 0 maps to red
[0.001, '#1f77b4'], # Slightly above 0 starts with a blue shade
[0.45, '#1f77b4'], # Keep blue until just before 46%
[0.46, 'red'], # 46% maps to red
[0.47, '#1f77b4'], # Back to blue after 46%
[0.49, '#1f77b4'], # Keep blue until just before 50%
[0.5, 'red'], # 50% maps to red
[0.51, '#aec7e8'], # Resume the original transition
[1, '#08306b'] # Dark blue at 1
]
fig = go.Figure(data=go.Heatmap(
x=heatmap_df['Date'],
y=heatmap_df['Trade'],
z=heatmap_df['Shortage_Probability'],
colorscale=custom_colorscale,
zmin=0, zmax=1,
text=display_probs.round(0).astype(int).astype(str) + '%',
texttemplate="%{text}",
textfont={"size": 14, "color": "black"},
hovertemplate="Trade: %{y}<br>Date: %{x}<br>Shortage Risk: %{text}<extra></extra>",
colorbar=dict(
title="Shortage Risk",
tickvals=[0, 0.5, 1],
ticktext=["0%", "50%", "100%"]
)
))
fig.update_layout(
title="Shortage Risk Heatmap",
xaxis_title="Date",
yaxis_title="Trade",
xaxis=dict(
tickangle=45,
tickformat="%Y-%m-%d",
showgrid=False
),
yaxis=dict(
autorange="reversed",
showgrid=False
),
font=dict(size=14),
margin=dict(l=100, r=50, t=100, b=100),
plot_bgcolor="white",
paper_bgcolor="white",
showlegend=False
)
return fig
except Exception as e:
logger.error(f"Error creating heatmap: {str(e)}")
return go.Figure().update_layout(title=f"Error in Heatmap: {str(e)}")
def create_chart(df, predictions_dict):
try:
combined_df = pd.DataFrame()
for trade, predictions in predictions_dict.items():
trade_df = df[df['Trade'].str.lower() == trade.lower()][['Date', 'Attendance']].copy()
trade_df['Type'] = 'Historical'
trade_df['Trade'] = trade
forecast_df = pd.DataFrame(predictions)
if not forecast_df.empty:
forecast_df['Date'] = pd.to_datetime(forecast_df['date'])
forecast_df['Attendance'] = forecast_df['headcount']
forecast_df['Type'] = 'Forecast'
forecast_df['Trade'] = trade
combined_df = pd.concat([combined_df, trade_df, forecast_df[['Date', 'Attendance', 'Type', 'Trade']]])
if combined_df.empty:
return go.Figure().update_layout(title="Labour Attendance Forecast (No Data)")
fig = px.line(
combined_df,
x='Date',
y='Attendance',
color='Trade',
line_dash='Type',
markers=True,
title='Labour Attendance Forecast by Trade'
)
return fig
except Exception as e:
logger.error(f"Error creating chart: {str(e)}")
return go.Figure().update_layout(title=f"Error in Chart: {str(e)}")
def generate_pdf_summary(trade_results):
try:
buffer = io.BytesIO()
with PdfPages(buffer) as pdf:
fig, ax = plt.subplots(figsize=(10, 6))
if not trade_results:
ax.text(0.1, 0.5, "No data available for summary", fontsize=12)
else:
for i, (trade, data) in enumerate(trade_results.items()):
ax.text(0.1, 0.9 - 0.1*i,
f"{trade}: {data['Attendance']} (Actual), Shortage Risk: {data['Shortage_risk'][0]*100:.0f}%", fontsize=12)
ax.set_title("Weekly Labour Forecast Summary")
ax.axis('off')
pdf.savefig()
plt.close()
pdf_base64 = base64.b64encode(buffer.getvalue()).decode()
logger.info("PDF summary generated")
return pdf_base64
except Exception as e:
logger.error(f"Error generating PDF: {str(e)}")
return None
def format_output(trade_results, site_calendar_date):
output_columns = Config.REQUIRED_COLUMNS + ['Forecast_Next_3_Days__c', 'Shortage_risk', 'Suggested_actions', 'Alert_status']
output = []
notifications = []
for trade, data in trade_results.items():
output.append(f"Trade: {trade}")
for key in output_columns:
if key == 'Date':
value = pd.to_datetime(site_calendar_date).strftime('%Y-%m-%d') if pd.notna(site_calendar_date) else 'N/A'
elif key == 'Forecast_Next_3_Days__c':
value = ', '.join([f"{item['date']}: {item['headcount']}" for item in data.get(key, [])]) if data.get(key) else 'N/A'
else:
value = data.get(key, 'N/A')
if key in ['Weather', 'Alert_status', 'Suggested_actions', 'Trade'] and value is not None:
value = str(value)
elif key == 'Shortage_risk' and value is not None:
value = str(round(value[0], 2))
elif key == 'Attendance' and value is not None:
value = str(int(value))
output.append(f" • {key}: {value}")
alert_status = data.get('Alert_status', 'Normal')
suggested_actions = data.get('Suggested_actions', 'Monitor')
if alert_status == 'Critical':
notification = f"Urgent Alert for {trade}: {suggested_actions} due to high shortage risk of {round(data.get('Shortage_risk', [0])[0] * 100)}%."
elif alert_status == 'Warning':
notification = f"Warning for {trade}: {suggested_actions} due to moderate shortage risk of {round(data.get('Shortage_risk', [0])[0] * 100)}%."
else:
notification = f"Notice for {trade}: {suggested_actions}, shortage risk is low at {round(data.get('Shortage_risk', [0])[0] * 100)}%."
notifications.append(notification)
output.append("")
formatted_output = "\n".join(output) if trade_results else "No valid trade data available."
formatted_notifications = "Contractor Notifications:\n" + "\n".join([f" • {notification}" for notification in notifications]) if notifications else "No notifications available."
return formatted_output, formatted_notifications
def push_to_salesforce(sf, trade_results, site_calendar_date):
try:
if sf is None:
return "Salesforce connection not established"
records_to_upsert = []
for trade, data in trade_results.items():
forecast_json = ', '.join([f"{item['date']}: {item['headcount']}" for item in data.get('Forecast_Next_3_Days__c', [])])
record = {
'Trade__c': trade,
'Date__c': site_calendar_date.strftime('%Y-%m-%d'),
'Expected_Headcount__c': int(data['Attendance']),
'Actual_Headcount__c': int(data['Attendance']),
'Forecast_Next_3_Days__c': forecast_json,
'Shortage_Risk__c': float(data['Shortage_risk'][0]),
'Suggested_Actions__c': str(data['Suggested_actions']),
'Alert_Status__c': str(data['Alert_status']),
'Dashboard_Display__c': True
}
records_to_upsert.append(record)
for record in records_to_upsert:
sf.Labour_Attendance_Forecast__c.create(record)
logger.info(f"Successfully pushed {len(records_to_upsert)} records to Salesforce")
return None
except Exception as e:
logger.error(f"Error pushing to Salesforce: {str(e)}")
return f"Error pushing to Salesforce: {str(e)}"
def generate_sample_csv():
sample_data = {
'Date': ['2025-06-12', '2025-06-12', '2025-06-12', '2025-06-12'],
'Attendance': [10, 15, 20, 12],
'Trade': ['Painter', 'Electrician', 'Carpenter', 'Plumber'],
'Weather': ['Sunny', 'Rainy', 'Cloudy', 'Sunny']
}
df = pd.DataFrame(sample_data)
buffer = io.StringIO()
df.to_csv(buffer, index=False, encoding='utf-8')
csv_base64 = base64.b64encode(buffer.getvalue().encode('utf-8')).decode()
return csv_base64
# Main forecast function
def forecast_labour(csv_file, trade_filter=None, site_calendar_date=None):
try:
logger.info("Starting forecast process")
if csv_file is None:
return "Error: No CSV file uploaded", None, None, None, "No notifications available."
# Validate site calendar date format
try:
if not site_calendar_date:
raise ValueError("Site calendar date is required")
logger.info(f"Raw site_calendar_date input: '{site_calendar_date}'")
site_calendar_date = site_calendar_date.strip()
try:
site_calendar_date = pd.to_datetime(site_calendar_date, format='%Y-%m-%d')
except ValueError as strict_error:
logger.warning(f"Strict date parsing failed: {str(strict_error)}. Attempting mixed format parsing.")
site_calendar_date = pd.to_datetime(site_calendar_date, format='mixed', dayfirst=True, errors='coerce')
if pd.isna(site_calendar_date):
raise ValueError("Invalid site calendar date format. Use YYYY-MM-DD (e.g., 2025-06-13)")
except ValueError as e:
logger.error(f"Date validation error: {str(e)}")
return f"Error: {str(e)}", None, None, None, "No notifications available."
logger.info(f"Processing CSV file: {csv_file}")
df, error = process_csv(csv_file)
if error:
return error, None, None, None, "No notifications available."
unique_trades = df['Trade'].dropna().unique()
logger.info(f"Unique trades in CSV: {list(unique_trades)}")
if trade_filter and trade_filter.strip():
selected_trades = [t.strip() for t in trade_filter.split(',') if t.strip()]
selected_trades = [t for t in selected_trades if any(t.lower() == ut.lower() for ut in unique_trades)]
if not selected_trades:
logger.warning(f"No valid trades found in filter: {trade_filter}. Defaulting to all trades.")
selected_trades = unique_trades
else:
logger.info("Trade filter empty. Using all trades.")
selected_trades = unique_trades
logger.info(f"Selected trades: {list(selected_trades)}")
trade_results = {}
predictions_dict = {}
shortage_probs_dict = {}
alert_statuses = {}
errors = []
for trade in selected_trades:
trade_df = df[df['Trade'].str.lower() == trade.lower()]
date_match = trade_df[trade_df['Date'] == site_calendar_date]
if date_match.empty:
errors.append(f"No data for trade {trade} on {site_calendar_date.strftime('%Y-%m-%d')}")
continue
if len(date_match) > 1:
errors.append(f"Warning: Multiple rows for trade {trade} on {site_calendar_date.strftime('%Y-%m-%d')}")
predictions, shortage_probs, site_calendar, suggested_actions, alert_status, forecast_error = weighted_moving_average_forecast(df, trade, site_calendar_date)
if forecast_error:
errors.append(forecast_error)
continue
predictions_dict[trade] = predictions
shortage_probs_dict[trade] = shortage_probs
alert_statuses[trade] = alert_status
record = date_match.iloc[0]
result_data = {
'Date': site_calendar_date,
'Trade': trade,
'Weather': record['Weather'],
'Attendance': record['Attendance'],
'Forecast_Next_3_Days__c': predictions,
'Shortage_risk': shortage_probs,
'Suggested_actions': suggested_actions,
'Alert_status': alert_status
}
trade_results[trade] = result_data
if not trade_results:
error_msg = "No valid trade data processed"
if errors:
error_msg += f". Errors: {'; '.join(errors)}"
return error_msg, None, None, None, "No notifications available."
sf = connect_to_salesforce()
sf_error = push_to_salesforce(sf, trade_results, site_calendar_date)
if sf_error:
errors.append(sf_error)
line_chart = create_chart(df, predictions_dict)
heatmap = create_heatmap(df, predictions_dict, shortage_probs_dict, site_calendar_date)
pdf_summary = generate_pdf_summary(trade_results)
formatted_output, formatted_notifications = format_output(trade_results, site_calendar_date)
error_msg = "; ".join(errors) if errors else None
final_output = formatted_output + (f"\nWarnings: {error_msg}" if error_msg else "")
return (
final_output,
line_chart,
heatmap,
f'<a href="data:application/pdf;base64,{pdf_summary}" download="summary.pdf">Download Summary PDF</a>',
formatted_notifications
)
except Exception as e:
logger.error(f"Unexpected error in forecast: {str(e)}", exc_info=True)
return f"Error processing file: {str(e)}", None, None, None, "No notifications available."
# Gradio interface
def gradio_interface():
sample_csv = generate_sample_csv()
with gr.Blocks(theme=gr.themes.Soft()) as interface:
gr.Markdown("# Labour Attendance Forecast")
gr.Markdown("Upload a CSV with columns: Date, Attendance, Trade, Weather")
gr.Markdown("Enter trade names (e.g., 'Painter, Electrician') or leave blank for all trades")
gr.Markdown("Enter site calendar date (YYYY-MM-DD) for CSV data and 3-day forecast")
gr.Markdown(f'<a href="data:text/csv;base64,{sample_csv}" download="sample_labour_data.csv">Download Sample CSV</a>')
with gr.Row():
csv_input = gr.File(label="Upload CSV", file_types=[".csv"])
trade_input = gr.Textbox(label="Filter by Trades", placeholder="e.g., Painter, Electrician")
site_calendar_input = gr.Textbox(label="Site Calendar Date (YYYY-MM-DD)", placeholder="e.g., 2025-06-13")
forecast_button = gr.Button("Generate Forecast")
result_output = gr.Textbox(label="Forecast Result", lines=20)
line_chart_output = gr.Plot(label="Forecast Trendline")
heatmap_output = gr.Plot(label="Real-Time Shortage Risk Heatmap")
notification_output = gr.Textbox(label="Contractor Notifications", lines=5)
pdf_output = gr.HTML(label="Download Summary PDF")
forecast_button.click(
fn=forecast_labour,
inputs=[csv_input, trade_input, site_calendar_input],
outputs=[result_output, line_chart_output, heatmap_output, pdf_output, notification_output]
)
logger.info("Launching Gradio interface")
return interface
if __name__ == '__main__':
interface = gradio_interface()
interface.launch(share=False)