import gradio as gr import numpy as np import pandas as pd import matplotlib.pyplot as plt from scipy import stats from rapidfuzz import fuzz ####################################### # Debug Logging Function ####################################### def debug_print(message): print(message) ####################################### # Data Generation Functions ####################################### def generate_case_data(num_records=5000): debug_print("Generating case data...") lob_list = np.random.choice(["Modern Life", "Xbox", "CAPS", "Devices", "Modern Work"], num_records) issue_types = np.random.choice(["Billing", "Technical", "Hacking", "Service", "Access"], num_records) advocates = np.random.choice(["Alice", "Bob", "Charlie", "Diana", "Eve"], num_records) start_date = pd.Timestamp("2021-01-01") end_date = pd.Timestamp("2023-12-31") days_range = (end_date - start_date).days # Generate random case dates over 3 years case_dates = start_date + pd.to_timedelta(np.random.randint(0, days_range, num_records), unit='D') # Simulated release dates per LOB (set in early 2022) lob_release_dates = { "Modern Life": pd.Timestamp("2022-01-01"), "Xbox": pd.Timestamp("2022-02-01"), "CAPS": pd.Timestamp("2022-03-01"), "Devices": pd.Timestamp("2022-04-01"), "Modern Work": pd.Timestamp("2022-05-01") } release_dates = np.array([lob_release_dates[lob] for lob in lob_list]) pre_release = case_dates < release_dates CSAT = np.where(pre_release, np.random.normal(80, 5, num_records), np.random.normal(85, 5, num_records)) days_to_close = np.where(pre_release, np.random.normal(5, 1, num_records), np.random.normal(4, 1, num_records)) first_contact_resolution = np.where(pre_release, np.random.normal(70, 8, num_records), np.random.normal(75, 8, num_records)) CPI = np.where(pre_release, np.random.normal(50, 5, num_records), np.random.normal(45, 5, num_records)) # For the main case dataset, we do NOT include initiative utilization columns. debug_print("Case data generated.") return pd.DataFrame({ "serial_number": np.arange(1, num_records + 1), "advocate": advocates, "LOB": lob_list, "issue_type": issue_types, "case_date": case_dates, "CSAT": CSAT, "days_to_close": days_to_close, "first_contact_resolution": first_contact_resolution, "CPI": CPI }) def generate_advocate_adoption_data(): debug_print("Generating advocate adoption data...") advocates = ["Alice", "Bob", "Charlie", "Diana", "Eve"] # Ensure adoption dates fall roughly in mid-2022 adoption_dates = { "Symbiosis_adoption_date": ["2022-06-05", "2022-06-10", "2022-06-08", "2022-06-12", "2022-06-07"], "Voice Translation_adoption_date": ["2022-06-03", "2022-06-07", "2022-06-05", "2022-06-09", "2022-06-04"], "NoteHero_adoption_date": ["2022-06-02", "2022-06-06", "2022-06-04", "2022-06-08", "2022-06-03"] } df = pd.DataFrame({ "advocate": advocates, "Symbiosis_adoption_date": pd.to_datetime(adoption_dates["Symbiosis_adoption_date"]), "Voice Translation_adoption_date": pd.to_datetime(adoption_dates["Voice Translation_adoption_date"]), "NoteHero_adoption_date": pd.to_datetime(adoption_dates["NoteHero_adoption_date"]) }) # Convert to date objects for col in ["Symbiosis_adoption_date", "Voice Translation_adoption_date", "NoteHero_adoption_date"]: df[col] = df[col].dt.date debug_print("Advocate adoption data generated.") return df def generate_utilization_data(): debug_print("Generating initiative utilization data...") # Create a standalone DataFrame with serial numbers and binary flags for each initiative. df = global_case_data.copy()[["serial_number"]].copy() # For demonstration, we simulate 50% usage for each initiative. for initiative in ["Voice Translation_utilized", "Symbiosis_utilized", "NoteHero_utilized"]: df[initiative] = np.random.choice([0, 1], size=len(df), p=[0.5, 0.5]) debug_print("Initiative utilization data generated.") return df ####################################### # Global Data Setup ####################################### global_case_data = generate_case_data(num_records=5000) global_advocate_adoption = generate_advocate_adoption_data() global_initiative_utilization = generate_utilization_data() # Ensure that the utilization dataset only contains serial numbers that are in the main dataset. valid_serials = set(global_case_data["serial_number"]) global_initiative_utilization = global_initiative_utilization[global_initiative_utilization["serial_number"].isin(valid_serials)] debug_print("Global datasets generated.") ####################################### # Helper Calculation Functions ####################################### def calculate_throughput(df, start_date, end_date): df_filtered = df.loc[(df["case_date"] >= start_date) & (df["case_date"] <= end_date)] num_cases = len(df_filtered) num_days = (end_date - start_date).days or 1 return num_cases / num_days if num_cases > 0 else 0 def calculate_throughput_per_advocate(df, start_date, end_date): df_filtered = df.loc[(df["case_date"] >= start_date) & (df["case_date"] <= end_date)] if df_filtered.empty: return None throughput = df_filtered.groupby(["LOB", "advocate"]).size() / (end_date - start_date).days return throughput ####################################### # Analysis Functions ####################################### def analyze_overall_impact(release_date_str, lob_filter, issue_filter, kpi, one_tailed): debug_print("Running Overall Impact Analysis...") try: # Parse release date as a date object (no .dt on a scalar) release_date = pd.to_datetime(release_date_str).date() except Exception as e: return f"Error parsing release date: {str(e)}", None df = global_case_data.copy() df["case_date"] = pd.to_datetime(df["case_date"]).dt.date if lob_filter != "All": df = df[df["LOB"] == lob_filter] if issue_filter != "All": df = df[df["issue_type"] == issue_filter] if df.empty: return "No data available for the selected filters.", None pre_data = df[df["case_date"] < release_date] post_data = df[df["case_date"] >= release_date] if pre_data.empty or post_data.empty: return "No data available for the selected date range.", None if kpi.lower() == "throughput": throughput_pre = calculate_throughput(pre_data, pre_data["case_date"].min(), pre_data["case_date"].max()) throughput_post = calculate_throughput(post_data, post_data["case_date"].min(), post_data["case_date"].max()) t_stat, p_value = stats.ttest_ind(np.array([throughput_pre]), np.array([throughput_post]), equal_var=False) else: pre_vals, post_vals = pre_data[kpi].values, post_data[kpi].values t_stat, p_value = stats.ttest_ind(pre_vals, post_vals, equal_var=False) if one_tailed: p_value = p_value / 2 significance = "Significant" if p_value < 0.05 and t_stat > 0 else "Not Significant" else: significance = "Significant" if p_value < 0.05 else "Not Significant" analysis_text = f"""Overall Impact Analysis for KPI: {kpi} Filters - LOB: {lob_filter}, Issue Type: {issue_filter} Global Release Date: {release_date} T-Test: T-Statistic = {t_stat:.3f}, P-Value = {p_value:.3f} ({significance}) """ # Here you could also add additional aggregated results if needed. fig, ax = plt.subplots(figsize=(6, 4)) if kpi.lower() == "throughput": # For throughput, show a simple bar graph with aggregated throughput (for demonstration) ax.bar(["Pre", "Post"], [throughput_pre, throughput_post], color=["blue", "green"]) ax.set_ylabel("Throughput (cases/day)") else: ax.boxplot([pre_data[kpi].values, post_data[kpi].values], labels=["Pre", "Post"]) ax.set_ylabel(kpi) ax.set_title("Overall Impact Analysis") plt.tight_layout() plt.close(fig) return analysis_text, fig def analyze_all_advocates_impact(method, initiative, lob_filter, issue_filter, kpi, one_tailed, adoption_file, adoption_name_col, adoption_date_col, utilization_file): try: debug_print("🚀 Running Advocate Impact Analysis...") df = global_case_data.copy() if lob_filter != "All": df = df[df["LOB"] == lob_filter] if issue_filter != "All": df = df[df["issue_type"] == issue_filter] if df.empty: debug_print("⚠ No cases available for the selected filters.") return "No data available for the selected filters.", None, None df["case_date"] = pd.to_datetime(df["case_date"], utc=True, errors="coerce").dt.normalize().dt.date debug_print(f"✅ Data filtered. {len(df)} cases remain.") debug_print(f"🗓 Min case date: {df['case_date'].min()}, Max case date: {df['case_date'].max()}") # For Initiative Utilization, use standalone DF utilization_df = global_initiative_utilization.copy() if method == "Initiative Utilization" and utilization_file is not None: try: util_df = pd.read_csv(utilization_file.name) except Exception: try: util_df = pd.read_excel(utilization_file.name) except Exception as e: debug_print(f"❌ Error reading utilization file: {str(e)}") return f"Error reading utilization file: {str(e)}", None, None if "serial_number" not in util_df.columns: debug_print("⚠ The uploaded utilization file must have a 'serial_number' column.") return "The uploaded utilization file must have a 'serial_number' column.", None, None utilization_df = util_df.copy() debug_print(f"✅ Uploaded initiative utilization file processed: {utilization_df.shape[0]} rows.") else: debug_print("📂 No initiative utilization file uploaded; using default global initiative utilization data.") # Build adoption mapping for Adoption Date method adoption_mapping = {} if method == "Adoption Date" and adoption_file is not None: try: uploaded_df = pd.read_csv(adoption_file.name) except Exception: try: uploaded_df = pd.read_excel(adoption_file.name) except Exception as e: debug_print(f"❌ Error reading adoption file: {str(e)}") return f"Error reading adoption file: {str(e)}", None, None if adoption_name_col not in uploaded_df.columns or adoption_date_col not in uploaded_df.columns: debug_print("⚠ Specified columns not found in the uploaded adoption file.") return "Specified columns not found in the uploaded adoption file.", None, None debug_print("📂 Processing uploaded adoption file...") for idx, row in uploaded_df.iterrows(): name_uploaded = str(row[adoption_name_col]) adoption_date = pd.to_datetime(row[adoption_date_col], utc=True, errors="coerce") if pd.isnull(adoption_date): debug_print(f"⚠ Skipping invalid adoption date for {name_uploaded}") continue adoption_date = adoption_date.date() # Map using fuzzy matching on the default global adoption names for adv in df["advocate"].unique(): score = fuzz.ratio(name_uploaded.lower(), adv.lower()) if score >= 95: adoption_mapping[adv] = min(adoption_mapping.get(adv, adoption_date), adoption_date) debug_print(f"✅ Uploaded adoption file processed. Mapped {len(adoption_mapping)} advocates.") else: debug_print("📂 No adoption file uploaded; using default global adoption data.") # Normalize global adoption dates for col in ["Symbiosis_adoption_date", "Voice Translation_adoption_date", "NoteHero_adoption_date"]: global_advocate_adoption[col] = pd.to_datetime(global_advocate_adoption[col], utc=True, errors="coerce") global_advocate_adoption[col] = global_advocate_adoption[col].apply(lambda x: x.date() if pd.notnull(x) else None) all_pre_vals, all_post_vals = [], [] results = [] debug_print("🔎 Processing advocates...") for adv in df["advocate"].unique(): try: df_adv = df[df["advocate"] == adv] if method == "Adoption Date": if adv in adoption_mapping: adoption_date = adoption_mapping[adv] else: col_name = initiative + "_adoption_date" adoption_series = global_advocate_adoption.loc[global_advocate_adoption["advocate"] == adv, col_name] if adoption_series.empty or pd.isnull(adoption_series.values[0]): debug_print(f"⚠ Skipping {adv}: No valid adoption date found.") continue adoption_date = adoption_series.values[0] if pd.isnull(adoption_date): debug_print(f"⚠ Skipping {adv}: Adoption date is NULL after conversion.") continue debug_print(f"🔎 Processing {adv}: Adoption Date = {adoption_date}") pre_data = df_adv[df_adv["case_date"] < adoption_date] post_data = df_adv[df_adv["case_date"] >= adoption_date] debug_print(f" {adv}: Pre-data count = {len(pre_data)}, Post-data count = {len(post_data)}") if pre_data.empty: debug_print(f"⚠ Skipping {adv}: No pre-adoption cases.") continue if post_data.empty: debug_print(f"⚠ Skipping {adv}: No post-adoption cases.") continue slice_info = f"Adoption Date: {adoption_date}" elif method == "Initiative Utilization": col_name = initiative + "_utilized" df_adv = df_adv.copy() df_adv = df_adv.merge(utilization_df[["serial_number", col_name]], on="serial_number", how="left") df_adv[col_name] = df_adv[col_name].fillna(0) pre_data = df_adv[df_adv[col_name] == 0] post_data = df_adv[df_adv[col_name] == 1] slice_info = "Initiative Utilization" else: continue if pre_data.empty or post_data.empty: debug_print(f"⚠ Advocate {adv}: Not enough data; skipping.") continue if kpi.lower() == "throughput": pre_val = calculate_throughput(pre_data, pre_data["case_date"].min(), pre_data["case_date"].max()) post_val = calculate_throughput(post_data, post_data["case_date"].min(), post_data["case_date"].max()) else: pre_val = np.mean(pre_data[kpi].values) post_val = np.mean(post_data[kpi].values) pct_change = ((post_val - pre_val) / pre_val) * 100 if pre_val else np.nan results.append({ "advocate": adv, "Pre_Mean": pre_val, "Post_Mean": post_val, "Percent_Change": pct_change, "Slice_Info": slice_info }) all_pre_vals.extend(pre_data[kpi].values) all_post_vals.extend(post_data[kpi].values) debug_print(f"✅ Processed {adv}: {pct_change:.2f}% change.") except Exception as e: debug_print(f"❌ Error processing {adv}: {str(e)}") if not results: debug_print("⚠ No valid advocates found for analysis.") return "No valid advocates found for analysis. Check the case date ranges.", None, None results_df = pd.DataFrame(results).sort_values(by="Percent_Change", ascending=False) # Perform aggregated T-Test try: if len(all_pre_vals) > 1 and len(all_post_vals) > 1: t_stat, p_value = stats.ttest_ind(all_pre_vals, all_post_vals, equal_var=False) if one_tailed: p_value = p_value / 2 significance = "Statistically Significant" if p_value < 0.05 and t_stat > 0 else "Not Statistically Significant" else: significance = "Statistically Significant" if p_value < 0.05 else "Not Statistically Significant" else: t_stat, p_value = np.nan, np.nan significance = "Insufficient Data for Statistical Test" except Exception as e: debug_print(f"❌ Error performing T-Test: {str(e)}") return f"Error performing T-Test: {str(e)}", None, None pre_mean = np.mean(all_pre_vals) if len(all_pre_vals) > 0 else np.nan post_mean = np.mean(all_post_vals) if len(all_post_vals) > 0 else np.nan overall_pct_change = ((post_mean - pre_mean) / pre_mean) * 100 if pre_mean else np.nan overall_summary = f"""📊 Aggregated Advocate Impact Analysis using method '{method}' for initiative '{initiative}' on KPI '{kpi}'. Number of advocates analyzed: {len(results_df)} Aggregated Pre vs Post Analysis: - Pre-Adoption Mean: {pre_mean:.2f} - Post-Adoption Mean: {post_mean:.2f} - Percent Change: {overall_pct_change:.2f}% T-Test Results: - T-Statistic: {t_stat:.3f} - P-Value: {p_value:.3f} - Result: {significance} """ fig, ax = plt.subplots(figsize=(6, 4)) ax.bar(["Pre-Adoption", "Post-Adoption"], [pre_mean, post_mean], color=["blue", "green"]) ax.set_title(f"Aggregated Impact of {initiative} on {kpi}") ax.set_ylabel(kpi) plt.tight_layout() plt.close(fig) debug_print("🎯 Advocate Impact Analysis completed.") return overall_summary, fig, results_df except Exception as e: debug_print(f"❌ Fatal Error in Function: {str(e)}") return f"Fatal Error: {str(e)}", None, None with gr.Blocks() as demo: gr.Markdown("# Impact Analysis Dashboard") with gr.Tabs(): # Tab 1: Overall Impact Analysis with gr.TabItem("Overall Impact Analysis"): gr.Markdown("### Overall Impact Analysis (Global Release Date)") overall_release_date = gr.Textbox(label="Global Release Date (YYYY-MM-DD)", placeholder="e.g., 2022-01-15") overall_lob = gr.Dropdown(choices=["All", "Modern Life", "Xbox", "CAPS", "Devices", "Modern Work"], label="Filter by LOB", value="All") overall_issue = gr.Dropdown(choices=["All", "Billing", "Technical", "Hacking", "Service", "Access"], label="Filter by Issue Type", value="All") overall_kpi = gr.Dropdown(choices=["CSAT", "days_to_close", "first_contact_resolution", "CPI", "throughput"], label="Select KPI", value="CSAT") one_tailed_overall = gr.Checkbox(label="Use One-Tailed T-Test") overall_btn = gr.Button("Analyze Overall Impact") overall_output = gr.Textbox(label="Overall Impact Analysis Results") overall_plot = gr.Plot(label="Overall Impact Graph") overall_btn.click(analyze_overall_impact, inputs=[overall_release_date, overall_lob, overall_issue, overall_kpi, one_tailed_overall], outputs=[overall_output, overall_plot]) # Tab 2: Advocate Impact Analysis with gr.TabItem("Advocate Impact Analysis"): gr.Markdown("### Advocate Impact Analysis (Aggregated Pre vs Post)") adoption_method = gr.Radio(choices=["Adoption Date", "Initiative Utilization"], label="Method", value="Adoption Date") initiative_select = gr.Dropdown(choices=["Symbiosis", "Voice Translation", "NoteHero"], label="Select Initiative", value="Symbiosis") adv_lob = gr.Dropdown(choices=["All", "Modern Life", "Xbox", "CAPS", "Devices", "Modern Work"], label="Filter by LOB", value="All") adv_issue = gr.Dropdown(choices=["All", "Billing", "Technical", "Hacking", "Service", "Access"], label="Filter by Issue Type", value="All") adv_kpi = gr.Dropdown(choices=["CSAT", "days_to_close", "first_contact_resolution", "CPI", "throughput"], label="Select KPI", value="CSAT") one_tailed_adv = gr.Checkbox(label="Use One-Tailed T-Test") with gr.Accordion("Optional File Uploads (Click to expand)", open=False): gr.Markdown("Upload is optional. For Adoption Date method, upload a CSV/Excel with two columns (Advocate Name and Adoption Date). For Initiative Utilization, upload a CSV/Excel with a 'serial_number' column.") adoption_file = gr.File(label="Upload Adoption Date File (optional)") adoption_name_col = gr.Textbox(label="Adoption File: Advocate Name Column", placeholder="e.g., Name") adoption_date_col = gr.Textbox(label="Adoption File: Adoption Date Column", placeholder="e.g., AdoptionDate") utilization_file = gr.File(label="Upload Initiative Utilization File (optional)") adv_btn = gr.Button("Analyze Advocate Impact") adv_overall_output = gr.Textbox(label="Aggregated Advocate Impact Summary") adv_plot = gr.Plot(label="Aggregated Advocate Impact Graph") adv_table = gr.Dataframe(label="Advocate Impact Details") adv_btn.click(analyze_all_advocates_impact, inputs=[adoption_method, initiative_select, adv_lob, adv_issue, adv_kpi, one_tailed_adv, adoption_file, adoption_name_col, adoption_date_col, utilization_file], outputs=[adv_overall_output, adv_plot, adv_table]) # Optional Debug Logs Tab with gr.TabItem("Debug Logs"): gr.Markdown("### Debug Logs") debug_btn = gr.Button("Refresh Debug Logs") debug_output = gr.Textbox(label="Debug Logs", lines=15) debug_btn.click(lambda: "Check console output for debug logs.", inputs=[], outputs=[debug_output]) demo.launch()