from Project import * from google_drive import * from notion import upload_to_notion from state import state import gradio as gr import json from prompt_configs import PROMPTS from typing import Any, Generator, Tuple import pandas as pd from io import StringIO def generate_step_1(progress=gr.Progress()) -> Generator[Tuple[str, list], Any, None]: """Generate content using the Project instance""" try: state.step1_results = [] for status_msg, results in state.quotation_project.generate_prd_and_components(progress): state.step1_results = results yield status_msg, results except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def generate_step_2a(progress=gr.Progress()): try: state.step2a_results = [] for status_msg , results , quotation_cost in state.quotation_project.generate_mandays_estimate(progress): state.step2a_results = results yield status_msg, results , quotation_cost except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def generate_step_2b(progress=gr.Progress()): try: state.step2b_results = [] for status_msg , results in state.quotation_project.analyze_mvp_components(progress): state.step2b_results = results yield status_msg, results except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def generate_step_2c(progress=gr.Progress()): try: state.step2c_results = [] for status_msg , results in state.quotation_project.recalculate_mvp_mandays(progress): state.step2c_results = results yield status_msg, results except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def generate_step_2d(progress=gr.Progress()): try: state.step2d_results = [] for status_msg , results , quotation_cost in state.quotation_project.generate_mvp_mandays(progress): state.step2d_results = results yield status_msg, results , quotation_cost except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def generate_step_3(progress=gr.Progress()): try: state.step3_results = [] for status_msg , results in state.quotation_project.generate_final_documentation(progress): state.step3_results = results yield status_msg, results except Exception as e: print(f"Error during generation: {str(e)}") yield f"Error during generation: {str(e)}", [] def recalculate_estimates(progress=gr.Progress()): """Handler for recalculate button click""" try: # Recalculate estimates updated_summary = state.quotation_project.recalculate_mandays_costs() return updated_summary except Exception as e: print(f"Error during recalculation: {str(e)}") return gr.Markdown(value=f"Error during recalculation: {str(e)}") def setup_all_handlers(step_buttons, all_components, progress_update, quotation_cost=None, recalc_btn=None, upload_drive_btn=None, upload_notion_btn=None, project_name=None, step1_results=None, step2a_results=None, step2b_results=None, step2c_results=None, step2d_results=None , step3_results=None): """Set up all step handlers with the provided UI components""" try: # Set up Step 1 handler step_buttons['Step 1 : Scope & Components'].click( fn=generate_step_1, outputs=[progress_update, step1_results], queue=True ) step_buttons['Step 2.1 : Generate Mandays'].click( fn=generate_step_2a, outputs=[progress_update, step2a_results , quotation_cost], queue=True ) step_buttons['Step 2.2 : Analyze MVP Components'].click( fn=generate_step_2b, outputs=[progress_update , step2b_results], queue=True ) step_buttons['Step 2.3 : Recalculate MVP Mandays'].click( fn=generate_step_2c, outputs=[progress_update , step2c_results], queue=True ) step_buttons['Step 2.4 : Generate MVP Mandays'].click( fn=generate_step_2d, outputs=[progress_update , step2d_results , quotation_cost], queue=True ) step_buttons['Step 3 : Final Documentation'].click( fn=generate_step_3, outputs=[progress_update , step3_results], queue=True ) recalc_btn.click( fn=recalculate_estimates, outputs=[quotation_cost], ) upload_drive_btn.click( fn=upload_to_gdrive, inputs=[project_name], outputs=[progress_update] ) upload_notion_btn.click( fn=upload_to_notion, inputs=[project_name], outputs=[progress_update] ) except Exception as e: print(f"Error setting up handlers: {str(e)}") return None def create_step1_render(step1_results): @gr.render(inputs=[step1_results]) def render_step1_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result = result_item.get("result") prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: text_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_text')), None) markdown_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_markdown')), None) # Create textbox and markdown components textbox = gr.Textbox( value=result, label=text_config.label, lines=10, interactive=True ) markdown = gr.Markdown( value=result, show_copy_button=True, label=markdown_config.label, elem_classes=["scrollable-markdown"] ) def update_project_attribute(new_text , fn_name = function_name): for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project , output , new_text) return new_text textbox.change(fn=update_project_attribute, inputs=[textbox] , outputs=[markdown]) result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step1_results def create_step2a_render(step2a_results): @gr.render(inputs=[step2a_results]) def render_step2a_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result_data = result_item.get("result", {}) prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: # Convert the data to pandas DataFrame first try: if isinstance(result_data, str) and ',' in result_data: # Handle CSV string clean_csv = result_data.replace('```csv', '').replace('```', '') df = pd.read_csv(StringIO(clean_csv)) else: # Handle dictionary data records = result_data.get(function_name, []) df = pd.DataFrame(records) if not df.empty: df_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_dataframe')), None) df_component = gr.DataFrame( value=df, label=df_config.label if df_config else f"{function_name} Results", interactive=df_config.interactive if df_config else False, wrap=True ) def update_project_attribute(updated_df, fn_name=function_name): try: # Convert DataFrame to dictionary format updated_records = updated_df.to_dict('records') updated_result = {fn_name: updated_records} for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project, output, updated_result) return updated_df except Exception as e: print(f"Error updating attribute: {str(e)}") return updated_df df_component.change( fn=update_project_attribute, inputs=[df_component], outputs=[df_component] ) else: gr.Markdown("No data available") except Exception as e: print(f"Error processing DataFrame: {str(e)}") gr.Markdown(f"Error creating DataFrame: {str(e)}") result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step2a_results def create_step2b_render(step2b_results): @gr.render(inputs=[step2b_results]) def render_step2b_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result_data = result_item.get("result", {}) prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: # Convert the data to pandas DataFrame first try: if isinstance(result_data, str) and ',' in result_data: # Handle CSV string clean_csv = result_data.replace('```csv', '').replace('```', '') df = pd.read_csv(StringIO(clean_csv)) else: # Handle dictionary data records = result_data.get(function_name, []) df = pd.DataFrame(records) if not df.empty: df_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_dataframe')), None) df_component = gr.DataFrame( value=df, label=df_config.label if df_config else f"{function_name} Results", interactive=df_config.interactive if df_config else False, wrap=True ) def update_project_attribute(updated_df, fn_name=function_name): try: # Convert DataFrame to dictionary format updated_records = updated_df.to_dict('records') updated_result = {fn_name: updated_records} for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project, output, updated_result) return updated_df except Exception as e: print(f"Error updating attribute: {str(e)}") return updated_df df_component.change( fn=update_project_attribute, inputs=[df_component], outputs=[df_component] ) else: gr.Markdown("No data available") except Exception as e: print(f"Error processing DataFrame: {str(e)}") gr.Markdown(f"Error creating DataFrame: {str(e)}") result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step2b_results def create_step2c_render(step2c_results): @gr.render(inputs=[step2c_results]) def render_step2c_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result = result_item.get("result") prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: text_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_text')), None) markdown_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_markdown')), None) # Create textbox and markdown components textbox = gr.Textbox( value=result, label=text_config.label, lines=10, interactive=True ) markdown = gr.Markdown( value=result, show_copy_button=True, label=markdown_config.label, elem_classes=["scrollable-markdown"] ) def update_project_attribute(new_text , fn_name = function_name): for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project , output , new_text) return new_text textbox.change(fn=update_project_attribute, inputs=[textbox] , outputs=[markdown]) result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step2c_results def create_step2d_render(step2d_results): @gr.render(inputs=[step2d_results]) def render_step2d_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result_data = result_item.get("result", {}) prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: try: # Handle each section separately for section_name, section_data in result_data.items(): if isinstance(section_data, list): df = pd.DataFrame(section_data) if not df.empty: # Get appropriate DataFrame config based on section df_config = None if 'MVP Plan_Test' in section_name: df_config = next((config for key, config in prompt_config.ui.items() if 'plan_test_mandays_dataframe' in key), None) elif 'MVP Dev' in section_name: df_config = next((config for key, config in prompt_config.ui.items() if 'dev_mandays_dataframe' in key), None) elif 'MVP Intents' in section_name: df_config = next((config for key, config in prompt_config.ui.items() if 'intents_mandays_dataframe' in key), None) df_component = gr.DataFrame( value=df, label=df_config.label if df_config else section_name, interactive=df_config.interactive if df_config else False, wrap=True ) def update_project_attribute(updated_df, fn_name=function_name, section=section_name): try: # Get current result data current_result = getattr(state.quotation_project, PROMPTS[fn_name].outputs[0], {}) if not isinstance(current_result, dict): current_result = {} # Update specific section current_result[section] = updated_df.to_dict('records') # Update project attribute for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project, output, current_result) return updated_df except Exception as e: print(f"Error updating attribute for {section}: {str(e)}") return updated_df df_component.change( fn=update_project_attribute, inputs=[df_component], outputs=[df_component] ) else: gr.Markdown(f"No data available for {section_name}") elif 'error' in section_name: gr.Markdown(f"Error in section: {section_data}") except Exception as e: print(f"Error processing DataFrames: {str(e)}") gr.Markdown(f"Error creating DataFrames: {str(e)}") result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step2d_results def create_step3_render(step3_results): @gr.render(inputs=[step3_results]) def render_step3_results(results): if not results: return [gr.Markdown("No results generated yet.")] result_components = [] for result_item in results: try: function_name = result_item.get("function_name") result = result_item.get("result") prompt_config = PROMPTS.get(function_name) if prompt_config: accordion = gr.Accordion( label=prompt_config.description, open=False ) with accordion: text_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_text')), None) markdown_config = next((config for key, config in prompt_config.ui.items() if key.endswith('_markdown')), None) # Create textbox and markdown components textbox = gr.Textbox( value=result, label=text_config.label, lines=10, interactive=True ) markdown = gr.Markdown( value=result, show_copy_button=True, label=markdown_config.label, elem_classes=["scrollable-markdown"] ) def update_project_attribute(new_text , fn_name = function_name): for output in PROMPTS[fn_name].outputs: setattr(state.quotation_project , output , new_text) return new_text textbox.change(fn=update_project_attribute, inputs=[textbox] , outputs=[markdown]) result_components.append(accordion) continue except Exception as e: print(f"Error processing result: {str(e)}") result_components.append( gr.Markdown(f"Error displaying result: {str(e)}") ) return result_components or [gr.Markdown("No valid results to display.")] return render_step3_results