# handlers/analytics_handlers.py import gradio as gr import logging import time from ui.analytics_plot_generator import update_analytics_plots_figures, create_placeholder_plot from ui.ui_generators import BOMB_ICON, EXPLORE_ICON, FORMULA_ICON, ACTIVE_ICON # Make sure these are accessible from features.chatbot.chatbot_prompts import get_initial_insight_prompt_and_suggestions from features.chatbot.chatbot_handler import generate_llm_response from config import PLOT_ID_TO_FORMULA_KEY_MAP # Ensure this is correctly imported from your config from formulas import PLOT_FORMULAS # Ensure this is correctly imported class AnalyticsHandlers: """Handles all analytics tab events and interactions.""" def __init__(self, analytics_components, token_state_ref, chat_histories_st_ref, current_chat_plot_id_st_ref, plot_data_for_chatbot_st_ref, active_panel_action_state_ref, explored_plot_id_state_ref): self.components = analytics_components self.plot_configs = analytics_components['plot_configs'] self.unique_ordered_sections = analytics_components['unique_ordered_sections'] self.num_unique_sections = len(self.unique_ordered_sections) self.plot_ui_objects = analytics_components['plot_ui_objects'] # e.g. {'plot_id1': {'panel_component': gr.Plot, 'bomb_button': gr.Button, ...}} self.section_titles_map = analytics_components['section_titles_map'] # e.g. {'Section Name': gr.Markdown} # References to global states, these are gr.State objects themselves self.token_state = token_state_ref self.chat_histories_st = chat_histories_st_ref self.current_chat_plot_id_st = current_chat_plot_id_st_ref self.plot_data_for_chatbot_st = plot_data_for_chatbot_st_ref self.active_panel_action_state = active_panel_action_state_ref self.explored_plot_id_state = explored_plot_id_state_ref logging.info(f"AnalyticsHandlers initialized. {len(self.plot_configs)} plot configs, {self.num_unique_sections} unique sections.") if not self.plot_ui_objects: logging.warning("AnalyticsHandlers: plot_ui_objects is empty or not correctly passed.") if not self.section_titles_map: logging.warning("AnalyticsHandlers: section_titles_map is empty or not correctly passed.") def _get_graph_refresh_outputs_list(self): """Helper to construct the list of outputs for graph refresh actions.""" outputs = [self.components['analytics_status_md']] # Plot components themselves for pc in self.plot_configs: plot_component = self.plot_ui_objects.get(pc["id"], {}).get("plot_component") if plot_component: outputs.append(plot_component) else: outputs.append(gr.update()) # Placeholder if not found logging.warning(f"Plot component for {pc['id']} not found in plot_ui_objects for refresh outputs.") # UI resets for action panel outputs.extend([ self.components['global_actions_column_ui'], self.components['insights_chatbot_ui'], # For value reset self.components['insights_chat_input_ui'], # For value reset self.components['insights_suggestions_row_ui'], self.components['insights_suggestion_1_btn'], self.components['insights_suggestion_2_btn'], self.components['insights_suggestion_3_btn'], self.components['formula_display_markdown_ui'], # For value reset self.components['formula_close_hint_md'] ]) # State resets outputs.extend([ self.active_panel_action_state, self.current_chat_plot_id_st, self.chat_histories_st, self.plot_data_for_chatbot_st ]) # Button and panel visibility resets for each plot for pc in self.plot_configs: plot_id = pc["id"] ui_obj = self.plot_ui_objects.get(plot_id, {}) outputs.extend([ ui_obj.get("bomb_button", gr.update()), ui_obj.get("formula_button", gr.update()), ui_obj.get("explore_button", gr.update()), ui_obj.get("panel_component", gr.update()) # For visibility reset ]) outputs.append(self.explored_plot_id_state) # Reset explored state # Section title visibility resets for s_name in self.unique_ordered_sections: outputs.append(self.section_titles_map.get(s_name, gr.update())) expected_len = 1 + len(self.plot_configs) + 9 + 4 + (4 * len(self.plot_configs)) + 1 + self.num_unique_sections # 1 (status) + N_plots (plots) + 9 (action panel UI) + 4 (states) + 4*N_plots (plot buttons/panels) + 1 (explored_id_state) + N_sections (titles) logging.debug(f"Graph refresh outputs list length: {len(outputs)}, Expected: {expected_len}") return outputs async def refresh_analytics_graphs_ui(self, current_token_state_val, date_filter_val, custom_start_val, custom_end_val, # chat_histories_st is a state, its value will be accessed via self.chat_histories_st.value ): logging.info(f"Refreshing analytics graph UI. Filter: {date_filter_val}. Token set: {'yes' if current_token_state_val.get('token') else 'no'}") start_time = time.time() # Call the function that generates plot figures and summaries # Ensure update_analytics_plots_figures is adapted to return: # status_msg (str), figures_dict (dict: {plot_id: fig}), summaries_dict (dict: {plot_id: summary_text}) plot_gen_results = update_analytics_plots_figures( current_token_state_val, date_filter_val, custom_start_val, custom_end_val, self.plot_configs # Pass plot_configs to it ) # Expected: status_msg, list_of_figures, dict_of_plot_summaries # Original: status_msg, gen_figs (list), new_summaries (dict) status_msg = plot_gen_results[0] gen_figs_list = plot_gen_results[1] # This should be a list of figures in order of plot_configs new_summaries_dict = plot_gen_results[2] # This should be a dict {plot_id: summary} all_updates = [gr.update(value=status_msg)] # For analytics_status_md # Update plot components with new figures if len(gen_figs_list) == len(self.plot_configs): for fig in gen_figs_list: all_updates.append(fig) # fig itself is the update for gr.Plot else: logging.error(f"Figure list length mismatch: got {len(gen_figs_list)}, expected {len(self.plot_configs)}") for _ in self.plot_configs: all_updates.append(create_placeholder_plot("Error", "Figura mancante")) # Reset action panel UI elements all_updates.extend([ gr.update(visible=False), # global_actions_column_ui gr.update(value=[], visible=False), # insights_chatbot_ui (value and visibility) gr.update(value="", visible=False), # insights_chat_input_ui (value and visibility) gr.update(visible=False), # insights_suggestions_row_ui gr.update(value="S1"), # insights_suggestion_1_btn gr.update(value="S2"), # insights_suggestion_2_btn gr.update(value="S3"), # insights_suggestion_3_btn gr.update(value="Formula details here.", visible=False), # formula_display_markdown_ui gr.update(visible=False) # formula_close_hint_md ]) # Reset states all_updates.extend([ None, # active_panel_action_state None, # current_chat_plot_id_st {}, # chat_histories_st (reset to empty dict) new_summaries_dict # plot_data_for_chatbot_st (update with new summaries) ]) # Reset buttons and panel visibility for each plot for _ in self.plot_configs: all_updates.extend([ gr.update(value=BOMB_ICON), # bomb_button gr.update(value=FORMULA_ICON), # formula_button gr.update(value=EXPLORE_ICON), # explore_button gr.update(visible=True) # panel_component (plot visibility) ]) all_updates.append(None) # explored_plot_id_state (reset) # Reset section title visibility for _ in self.unique_ordered_sections: all_updates.append(gr.update(visible=True)) end_time = time.time() logging.info(f"Analytics graph refresh processing took {end_time - start_time:.2f} seconds.") expected_len = 1 + len(self.plot_configs) + 9 + 4 + (4 * len(self.plot_configs)) + 1 + self.num_unique_sections logging.info(f"Prepared {len(all_updates)} updates for graph refresh. Expected {expected_len}.") if len(all_updates) != expected_len: logging.error(f"Output length mismatch in refresh_analytics_graphs_ui. Got {len(all_updates)}, expected {expected_len}") # Pad with gr.update() if lengths don't match, to avoid Gradio errors, though this indicates a logic flaw. all_updates.extend([gr.update()] * (expected_len - len(all_updates))) return tuple(all_updates) def _get_action_panel_outputs_list(self): """Helper to construct the list of outputs for panel actions (insights, formula).""" outputs = [ self.components['global_actions_column_ui'], self.components['insights_chatbot_ui'], # For visibility self.components['insights_chatbot_ui'], # For value self.components['insights_chat_input_ui'], self.components['insights_suggestions_row_ui'], self.components['insights_suggestion_1_btn'], self.components['insights_suggestion_2_btn'], self.components['insights_suggestion_3_btn'], self.components['formula_display_markdown_ui'], # For visibility self.components['formula_display_markdown_ui'], # For value self.components['formula_close_hint_md'], ] outputs.extend([ self.active_panel_action_state, self.current_chat_plot_id_st, self.chat_histories_st, self.explored_plot_id_state ]) for pc in self.plot_configs: ui_obj = self.plot_ui_objects.get(pc["id"], {}) outputs.append(ui_obj.get("panel_component", gr.update())) # Plot panel visibility outputs.append(ui_obj.get("bomb_button", gr.update())) outputs.append(ui_obj.get("formula_button", gr.update())) outputs.append(ui_obj.get("explore_button", gr.update())) for s_name in self.unique_ordered_sections: outputs.append(self.section_titles_map.get(s_name, gr.update())) # Section title visibility expected_len = 11 + 4 + (4 * len(self.plot_configs)) + self.num_unique_sections logging.debug(f"Action panel outputs list length: {len(outputs)}, Expected: {expected_len}") return outputs async def handle_panel_action(self, plot_id_clicked: str, action_type: str, current_active_action_from_state: dict, # This is a direct value from gr.State current_chat_histories: dict, # This is a direct value current_chat_plot_id: str, # This is a direct value current_plot_data_for_chatbot: dict, # This is a direct value current_explored_plot_id: str # This is a direct value ): logging.info(f"Panel Action: '{action_type}' for plot '{plot_id_clicked}'. Active: {current_active_action_from_state}, Explored: {current_explored_plot_id}") clicked_plot_config = next((p for p in self.plot_configs if p["id"] == plot_id_clicked), None) if not clicked_plot_config: logging.error(f"Config not found for plot_id {plot_id_clicked}") # Construct a list of gr.update() of the correct length num_outputs = len(self._get_action_panel_outputs_list()) error_updates = [gr.update()] * num_outputs # Try to preserve existing state values if possible by updating specific indices # This part is tricky without knowing the exact order and meaning of each output. # For simplicity, returning all gr.update() might be safer if an error occurs early. # Or, more robustly, identify which states need to be passed through. # Indices for states in action_panel_outputs_list: # active_panel_action_state is at index 11 # current_chat_plot_id_st is at index 12 # chat_histories_st is at index 13 # explored_plot_id_state is at index 14 error_updates[11] = current_active_action_from_state error_updates[12] = current_chat_plot_id error_updates[13] = current_chat_histories error_updates[14] = current_explored_plot_id return tuple(error_updates) clicked_plot_label = clicked_plot_config["label"] clicked_plot_section = clicked_plot_config["section"] hypothetical_new_active_state = {"plot_id": plot_id_clicked, "type": action_type} is_toggling_off = current_active_action_from_state == hypothetical_new_active_state action_col_visible_update = gr.update(visible=False) insights_chatbot_visible_update = gr.update(visible=False) insights_chat_input_visible_update = gr.update(visible=False) insights_suggestions_row_visible_update = gr.update(visible=False) formula_display_visible_update = gr.update(visible=False) formula_close_hint_visible_update = gr.update(visible=False) chatbot_content_update = gr.update() s1_upd, s2_upd, s3_upd = gr.update(), gr.update(), gr.update() formula_content_update = gr.update() new_active_action_state_to_set = None # This will be the new value for the gr.State new_current_chat_plot_id = current_chat_plot_id # Default to existing updated_chat_histories = current_chat_histories # Default to existing new_explored_plot_id_to_set = current_explored_plot_id # Default to existing generated_panel_vis_updates = [] # For individual plot panels generated_bomb_btn_updates = [] generated_formula_btn_updates = [] generated_explore_btn_updates = [] section_title_vis_updates = [gr.update()] * self.num_unique_sections if is_toggling_off: new_active_action_state_to_set = None action_col_visible_update = gr.update(visible=False) logging.info(f"Toggling OFF panel {action_type} for {plot_id_clicked}.") for _ in self.plot_configs: generated_bomb_btn_updates.append(gr.update(value=BOMB_ICON)) generated_formula_btn_updates.append(gr.update(value=FORMULA_ICON)) if current_explored_plot_id: # If an explore view is active, restore it explored_cfg = next((p for p in self.plot_configs if p["id"] == current_explored_plot_id), None) explored_sec = explored_cfg["section"] if explored_cfg else None for i, sec_name in enumerate(self.unique_ordered_sections): section_title_vis_updates[i] = gr.update(visible=(sec_name == explored_sec)) for cfg in self.plot_configs: is_exp = (cfg["id"] == current_explored_plot_id) generated_panel_vis_updates.append(gr.update(visible=is_exp)) generated_explore_btn_updates.append(gr.update(value=ACTIVE_ICON if is_exp else EXPLORE_ICON)) else: # No explore view, all plots/sections visible for i in range(self.num_unique_sections): section_title_vis_updates[i] = gr.update(visible=True) for _ in self.plot_configs: generated_panel_vis_updates.append(gr.update(visible=True)) generated_explore_btn_updates.append(gr.update(value=EXPLORE_ICON)) if action_type == "insights": new_current_chat_plot_id = None # Clear chat context if insights panel is closed else: # Toggling ON a new action or switching actions new_active_action_state_to_set = hypothetical_new_active_state action_col_visible_update = gr.update(visible=True) new_explored_plot_id_to_set = None # Cancel any explore view logging.info(f"Toggling ON panel {action_type} for {plot_id_clicked}. Cancelling explore view if any.") # Show only the section of the clicked plot for i, sec_name in enumerate(self.unique_ordered_sections): section_title_vis_updates[i] = gr.update(visible=(sec_name == clicked_plot_section)) # Show only the clicked plot's panel, update explore buttons to non-active for cfg in self.plot_configs: generated_panel_vis_updates.append(gr.update(visible=(cfg["id"] == plot_id_clicked))) generated_explore_btn_updates.append(gr.update(value=EXPLORE_ICON)) # Reset all explore to inactive # Update bomb and formula buttons based on the new active action for cfg_btn in self.plot_configs: is_active_insights = (new_active_action_state_to_set["plot_id"] == cfg_btn["id"] and new_active_action_state_to_set["type"] == "insights") is_active_formula = (new_active_action_state_to_set["plot_id"] == cfg_btn["id"] and new_active_action_state_to_set["type"] == "formula") generated_bomb_btn_updates.append(gr.update(value=ACTIVE_ICON if is_active_insights else BOMB_ICON)) generated_formula_btn_updates.append(gr.update(value=ACTIVE_ICON if is_active_formula else FORMULA_ICON)) if action_type == "insights": insights_chatbot_visible_update = gr.update(visible=True) insights_chat_input_visible_update = gr.update(visible=True) insights_suggestions_row_visible_update = gr.update(visible=True) new_current_chat_plot_id = plot_id_clicked # Set chat context history = current_chat_histories.get(plot_id_clicked, []) summary_for_plot = current_plot_data_for_chatbot.get(plot_id_clicked, f"Nessun sommario disponibile per '{clicked_plot_label}'.") if not history: # First time opening insights for this plot (or after a refresh) prompt, sugg = get_initial_insight_prompt_and_suggestions(plot_id_clicked, clicked_plot_label, summary_for_plot) # Gradio's chatbot expects a list of lists/tuples: [[user_msg, None], [None, assistant_msg]] # Our generate_llm_response and history uses: [{"role": "user", "content": prompt}, {"role": "assistant", "content": resp}] # We need to adapt. For now, let's assume generate_llm_response takes our format and returns a string. # The history for Gradio Chatbot component needs to be [[user_msg, assistant_msg], ...] # Let's build history for LLM first llm_history_for_generation = [{"role": "user", "content": prompt}] # Display "Thinking..." or similar chatbot_content_update = gr.update(value=[[prompt, "Sto pensando..."]]) yield tuple(self._assemble_panel_action_updates(action_col_visible_update, insights_chatbot_visible_update, chatbot_content_update, insights_chat_input_visible_update, insights_suggestions_row_visible_update, s1_upd, s2_upd, s3_upd, formula_display_visible_update, formula_content_update, formula_close_hint_visible_update, new_active_action_state_to_set, new_current_chat_plot_id, updated_chat_histories, new_explored_plot_id_to_set, generated_panel_vis_updates, generated_bomb_btn_updates, generated_formula_btn_updates, generated_explore_btn_updates, section_title_vis_updates)) resp_text = await generate_llm_response(prompt, plot_id_clicked, clicked_plot_label, llm_history_for_generation, summary_for_plot) # Gradio chatbot history format new_gr_history_for_plot = [[prompt, resp_text]] # Internal history format for re-sending to LLM new_internal_history_for_plot = [ {"role": "user", "content": prompt}, {"role": "assistant", "content": resp_text} ] updated_chat_histories = {**current_chat_histories, plot_id_clicked: new_internal_history_for_plot} chatbot_content_update = gr.update(value=new_gr_history_for_plot) else: # History exists, just display it _, sugg = get_initial_insight_prompt_and_suggestions(plot_id_clicked, clicked_plot_label, summary_for_plot) # Get fresh suggestions # Convert internal history to Gradio format for display gr_history_to_display = [] # Assuming history is [{"role":"user", "content":"..."}, {"role":"assistant", "content":"..."}] # We need to pair them up. If an odd number, the last user message might not have a pair yet. temp_hist = history[:] # Make a copy while temp_hist: user_turn = temp_hist.pop(0) assistant_turn = None if temp_hist and temp_hist[0]["role"] == "assistant": assistant_turn = temp_hist.pop(0) gr_history_to_display.append([user_turn["content"], assistant_turn["content"] if assistant_turn else None]) chatbot_content_update = gr.update(value=gr_history_to_display) s1_upd = gr.update(value=sugg[0] if sugg and len(sugg) > 0 else "N/A") s2_upd = gr.update(value=sugg[1] if sugg and len(sugg) > 1 else "N/A") s3_upd = gr.update(value=sugg[2] if sugg and len(sugg) > 2 else "N/A") elif action_type == "formula": formula_display_visible_update = gr.update(visible=True) formula_close_hint_visible_update = gr.update(visible=True) formula_key = PLOT_ID_TO_FORMULA_KEY_MAP.get(plot_id_clicked) formula_text = f"**Formula/Methodology for: {clicked_plot_label}** (ID: `{plot_id_clicked}`)\n\n" if formula_key and formula_key in PLOT_FORMULAS: formula_data = PLOT_FORMULAS[formula_key] formula_text += f"### {formula_data['title']}\n\n{formula_data['description']}\n\n" if 'calculation_steps' in formula_data and formula_data['calculation_steps']: formula_text += "**Calculation:**\n" + "\n".join([f"- {s}" for s in formula_data['calculation_steps']]) else: formula_text += "(No detailed formula information found.)" formula_content_update = gr.update(value=formula_text) new_current_chat_plot_id = None # Clear chat context if formula panel is opened final_updates_tuple = self._assemble_panel_action_updates( action_col_visible_update, insights_chatbot_visible_update, chatbot_content_update, insights_chat_input_visible_update, insights_suggestions_row_visible_update, s1_upd, s2_upd, s3_upd, formula_display_visible_update, formula_content_update, formula_close_hint_visible_update, new_active_action_state_to_set, new_current_chat_plot_id, updated_chat_histories, new_explored_plot_id_to_set, generated_panel_vis_updates, generated_bomb_btn_updates, generated_formula_btn_updates, generated_explore_btn_updates, section_title_vis_updates ) logging.debug(f"handle_panel_action returning {len(final_updates_tuple)} updates.") yield final_updates_tuple def _assemble_panel_action_updates(self, action_col_visible_update, insights_chatbot_visible_update, chatbot_content_update, insights_chat_input_visible_update, insights_suggestions_row_visible_update, s1_upd, s2_upd, s3_upd, formula_display_visible_update, formula_content_update, formula_close_hint_visible_update, new_active_action_state_to_set, new_current_chat_plot_id, updated_chat_histories, new_explored_plot_id_to_set, generated_panel_vis_updates, generated_bomb_btn_updates, generated_formula_btn_updates, generated_explore_btn_updates, section_title_vis_updates): """Helper to assemble the final tuple of updates for handle_panel_action.""" final_updates_list = [ action_col_visible_update, # global_actions_column_ui (visibility) insights_chatbot_visible_update, # insights_chatbot_ui (visibility) chatbot_content_update, # insights_chatbot_ui (value) insights_chat_input_visible_update, # insights_chat_input_ui insights_suggestions_row_visible_update, # insights_suggestions_row_ui s1_upd, # insights_suggestion_1_btn s2_upd, # insights_suggestion_2_btn s3_upd, # insights_suggestion_3_btn formula_display_visible_update, # formula_display_markdown_ui (visibility) formula_content_update, # formula_display_markdown_ui (value) formula_close_hint_visible_update, # formula_close_hint_md # States new_active_action_state_to_set, # active_panel_action_state new_current_chat_plot_id, # current_chat_plot_id_st updated_chat_histories, # chat_histories_st new_explored_plot_id_to_set # explored_plot_id_state ] final_updates_list.extend(generated_panel_vis_updates) final_updates_list.extend(generated_bomb_btn_updates) final_updates_list.extend(generated_formula_btn_updates) final_updates_list.extend(generated_explore_btn_updates) final_updates_list.extend(section_title_vis_updates) expected_len = len(self._get_action_panel_outputs_list()) if len(final_updates_list) != expected_len: logging.error(f"Output length mismatch in _assemble_panel_action_updates. Got {len(final_updates_list)}, expected {expected_len}") # Pad if necessary, though this is a bug indicator final_updates_list.extend([gr.update()] * (expected_len - len(final_updates_list))) return tuple(final_updates_list) async def handle_chat_message_submission(self, user_message: str, current_plot_id: str, chat_histories: dict, current_plot_data_for_chatbot: dict): if not current_plot_id or not user_message.strip(): # Get current Gradio history for the plot_id to display internal_history_for_plot = chat_histories.get(current_plot_id, []) gr_history_display = self._convert_internal_to_gradio_chat_history(internal_history_for_plot) yield gr_history_display, gr.update(value=""), chat_histories return clicked_plot_config = next((p for p in self.plot_configs if p["id"] == current_plot_id), None) plot_label = clicked_plot_config["label"] if clicked_plot_config else "Selected Plot" summary_for_plot = current_plot_data_for_chatbot.get(current_plot_id, f"No summary for '{plot_label}'.") internal_history_for_plot = chat_histories.get(current_plot_id, []).copy() # Get a mutable copy internal_history_for_plot.append({"role": "user", "content": user_message}) # Update Gradio chat display: User message + "Thinking..." gr_history_display_pending = self._convert_internal_to_gradio_chat_history(internal_history_for_plot, thinking=True) yield gr_history_display_pending, gr.update(value=""), chat_histories # Show user message immediately # Generate LLM response llm_response_text = await generate_llm_response(user_message, current_plot_id, plot_label, internal_history_for_plot, summary_for_plot) internal_history_for_plot.append({"role": "assistant", "content": llm_response_text}) updated_chat_histories = {**chat_histories, current_plot_id: internal_history_for_plot} # Final Gradio chat display with LLM response final_gr_history_display = self._convert_internal_to_gradio_chat_history(internal_history_for_plot) yield final_gr_history_display, "", updated_chat_histories def _convert_internal_to_gradio_chat_history(self, internal_history, thinking=False): """Converts internal chat history format to Gradio's [[user, assistant], ...] format.""" gradio_history = [] temp_hist = internal_history[:] # Make a copy while temp_hist: user_msg_obj = temp_hist.pop(0) user_msg = user_msg_obj['content'] assistant_msg = None if temp_hist and temp_hist[0]['role'] == 'assistant': assistant_msg_obj = temp_hist.pop(0) assistant_msg = assistant_msg_obj['content'] gradio_history.append([user_msg, assistant_msg]) if thinking and gradio_history and gradio_history[-1][1] is None: # If last message was user and we are in 'thinking' mode gradio_history[-1][1] = "Sto pensando..." # Replace None with "Thinking..." elif thinking and not gradio_history: # Should not happen if user_message was added pass return gradio_history async def handle_suggested_question_click(self, suggestion_text: str, current_plot_id: str, chat_histories: dict, current_plot_data_for_chatbot: dict): if not current_plot_id or not suggestion_text.strip() or suggestion_text == "N/A": internal_history_for_plot = chat_histories.get(current_plot_id, []) gr_history_display = self._convert_internal_to_gradio_chat_history(internal_history_for_plot) yield gr_history_display, gr.update(value=""), chat_histories return # Use the existing chat submission logic async for update_chunk in self.handle_chat_message_submission(suggestion_text, current_plot_id, chat_histories, current_plot_data_for_chatbot): yield update_chunk def _get_explore_outputs_list(self): """Helper to construct the list of outputs for explore actions.""" outputs = [ self.explored_plot_id_state, self.components['global_actions_column_ui'], # For visibility self.active_panel_action_state, # To potentially clear it self.components['formula_close_hint_md'] # For visibility ] for pc in self.plot_configs: # Plot panel visibility outputs.append(self.plot_ui_objects.get(pc["id"], {}).get("panel_component", gr.update())) for pc in self.plot_configs: # Explore button state outputs.append(self.plot_ui_objects.get(pc["id"], {}).get("explore_button", gr.update())) for pc in self.plot_configs: # Bomb button state (may need reset) outputs.append(self.plot_ui_objects.get(pc["id"], {}).get("bomb_button", gr.update())) for pc in self.plot_configs: # Formula button state (may need reset) outputs.append(self.plot_ui_objects.get(pc["id"], {}).get("formula_button", gr.update())) for s_name in self.unique_ordered_sections: # Section title visibility outputs.append(self.section_titles_map.get(s_name, gr.update())) expected_len = 4 + (4 * len(self.plot_configs)) + self.num_unique_sections logging.debug(f"Explore outputs list length: {len(outputs)}, Expected: {expected_len}") return outputs def handle_explore_click(self, plot_id_clicked: str, current_explored_plot_id_from_state: str, current_active_panel_action_state: dict): logging.info(f"Explore Click: Plot '{plot_id_clicked}'. Current Explored: {current_explored_plot_id_from_state}. Active Panel: {current_active_panel_action_state}") if not self.plot_ui_objects or not self.section_titles_map: logging.error("plot_ui_objects or section_titles_map not populated for handle_explore_click.") num_outputs = len(self._get_explore_outputs_list()) error_updates = [gr.update()] * num_outputs error_updates[0] = current_explored_plot_id_from_state # Preserve explored_id_state error_updates[2] = current_active_panel_action_state # Preserve active_panel_state return tuple(error_updates) new_explored_id_to_set = None is_toggling_off_explore = (plot_id_clicked == current_explored_plot_id_from_state) action_col_upd = gr.update() # Default no change new_active_panel_state_upd = current_active_panel_action_state # Default no change formula_hint_upd = gr.update(visible=False) # Default hide panel_vis_updates = [] explore_btns_updates = [] bomb_btns_updates = [gr.update()] * len(self.plot_configs) # Default no change formula_btns_updates = [gr.update()] * len(self.plot_configs) # Default no change section_title_vis_updates = [gr.update()] * self.num_unique_sections clicked_cfg = next((p for p in self.plot_configs if p["id"] == plot_id_clicked), None) section_of_clicked_plot = clicked_cfg["section"] if clicked_cfg else None if is_toggling_off_explore: new_explored_id_to_set = None # Clear explore state logging.info(f"Stopping explore for {plot_id_clicked}. All plots/sections to be visible.") for i in range(self.num_unique_sections): section_title_vis_updates[i] = gr.update(visible=True) for _ in self.plot_configs: panel_vis_updates.append(gr.update(visible=True)) explore_btns_updates.append(gr.update(value=EXPLORE_ICON)) # Bomb and formula buttons remain as they were unless an action panel was closed (handled below if current_active_panel_action_state was set) else: # Starting explore or switching explored plot new_explored_id_to_set = plot_id_clicked logging.info(f"Exploring {plot_id_clicked}. Hiding other plots/sections.") for i, sec_name in enumerate(self.unique_ordered_sections): section_title_vis_updates[i] = gr.update(visible=(sec_name == section_of_clicked_plot)) for cfg in self.plot_configs: is_target = (cfg["id"] == new_explored_id_to_set) panel_vis_updates.append(gr.update(visible=is_target)) explore_btns_updates.append(gr.update(value=ACTIVE_ICON if is_target else EXPLORE_ICON)) if current_active_panel_action_state: # If an action panel (insights/formula) is open, close it logging.info("Closing active insight/formula panel due to explore click.") action_col_upd = gr.update(visible=False) new_active_panel_state_upd = None # Clear active panel state formula_hint_upd = gr.update(visible=False) # Hide formula hint specifically # Reset bomb and formula buttons to their default icons bomb_btns_updates = [gr.update(value=BOMB_ICON) for _ in self.plot_configs] formula_btns_updates = [gr.update(value=FORMULA_ICON) for _ in self.plot_configs] final_explore_updates_list = [ new_explored_id_to_set, action_col_upd, new_active_panel_state_upd, formula_hint_upd ] final_explore_updates_list.extend(panel_vis_updates) final_explore_updates_list.extend(explore_btns_updates) final_explore_updates_list.extend(bomb_btns_updates) final_explore_updates_list.extend(formula_btns_updates) final_explore_updates_list.extend(section_title_vis_updates) expected_len = len(self._get_explore_outputs_list()) if len(final_explore_updates_list) != expected_len: logging.error(f"Output length mismatch in handle_explore_click. Got {len(final_explore_updates_list)}, expected {expected_len}") final_explore_updates_list.extend([gr.update()] * (expected_len - len(final_explore_updates_list))) return tuple(final_explore_updates_list) def setup_event_handlers(self): """Set up all event handlers for the analytics tab components.""" logging.info("Setting up analytics event handlers.") # Apply filter button apply_filter_inputs = [ self.token_state, self.components['date_filter_selector'], self.components['custom_start_date_picker'], self.components['custom_end_date_picker'], # self.chat_histories_st # Not directly an input to refresh_analytics_graphs_ui, it's accessed via self ] self.components['apply_filter_btn'].click( fn=self.refresh_analytics_graphs_ui, inputs=apply_filter_inputs, outputs=self._get_graph_refresh_outputs_list(), # Method returns the list of components show_progress="full", api_name="refresh_analytics_graphs" ) # Plot action handlers (insights, formula, explore) action_click_inputs = [ # These are the gr.State objects themselves self.active_panel_action_state, self.chat_histories_st, self.current_chat_plot_id_st, self.plot_data_for_chatbot_st, self.explored_plot_id_state ] explore_click_inputs = [ # gr.State objects self.explored_plot_id_state, self.active_panel_action_state ] action_panel_outputs_list = self._get_action_panel_outputs_list() explore_outputs_list = self._get_explore_outputs_list() for config_item in self.plot_configs: plot_id = config_item["id"] if plot_id in self.plot_ui_objects: ui_obj = self.plot_ui_objects[plot_id] # Curry plot_id and action_type for the handler # The handler function itself (self.handle_panel_action) will receive the values from the gr.State inputs directly. if ui_obj.get("bomb_button"): ui_obj["bomb_button"].click( fn=lambda current_active, current_chats, current_chat_pid, current_plot_data, current_explored, p_id=plot_id: \ self.handle_panel_action(p_id, "insights", current_active, current_chats, current_chat_pid, current_plot_data, current_explored), inputs=action_click_inputs, # Pass the list of gr.State objects outputs=action_panel_outputs_list, api_name=f"action_insights_{plot_id}" ) if ui_obj.get("formula_button"): ui_obj["formula_button"].click( fn=lambda current_active, current_chats, current_chat_pid, current_plot_data, current_explored, p_id=plot_id: \ self.handle_panel_action(p_id, "formula", current_active, current_chats, current_chat_pid, current_plot_data, current_explored), inputs=action_click_inputs, outputs=action_panel_outputs_list, api_name=f"action_formula_{plot_id}" ) if ui_obj.get("explore_button"): ui_obj["explore_button"].click( fn=lambda current_explored_val, current_active_panel_val, p_id=plot_id: \ self.handle_explore_click(p_id, current_explored_val, current_active_panel_val), inputs=explore_click_inputs, # Pass the list of gr.State objects outputs=explore_outputs_list, api_name=f"action_explore_{plot_id}" ) else: logging.warning(f"UI object for plot_id '{plot_id}' not found for setting up click handlers.") # Chat submission handlers chat_submission_outputs = [ self.components['insights_chatbot_ui'], self.components['insights_chat_input_ui'], self.chat_histories_st # This state will be updated ] chat_submission_inputs = [ # gr.Textbox, gr.State, gr.State, gr.State self.components['insights_chat_input_ui'], self.current_chat_plot_id_st, self.chat_histories_st, self.plot_data_for_chatbot_st ] self.components['insights_chat_input_ui'].submit( fn=self.handle_chat_message_submission, inputs=chat_submission_inputs, outputs=chat_submission_outputs, api_name="submit_chat_message" ) suggestion_click_inputs_base = [ # gr.State, gr.State, gr.State self.current_chat_plot_id_st, self.chat_histories_st, self.plot_data_for_chatbot_st ] # For suggestion buttons, the first input is the button itself (to get its value) self.components['insights_suggestion_1_btn'].click( fn=self.handle_suggested_question_click, inputs=[self.components['insights_suggestion_1_btn']] + suggestion_click_inputs_base, outputs=chat_submission_outputs, api_name="click_suggestion_1" ) self.components['insights_suggestion_2_btn'].click( fn=self.handle_suggested_question_click, inputs=[self.components['insights_suggestion_2_btn']] + suggestion_click_inputs_base, outputs=chat_submission_outputs, api_name="click_suggestion_2" ) self.components['insights_suggestion_3_btn'].click( fn=self.handle_suggested_question_click, inputs=[self.components['insights_suggestion_3_btn']] + suggestion_click_inputs_base, outputs=chat_submission_outputs, api_name="click_suggestion_3" ) logging.info("Analytics event handlers setup complete.")