Spaces:
Sleeping
Sleeping
| 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) |