from typing import Dict, Any, List import logging import pandas as pd import streamlit as st from utility_classes.base_report_page import BaseReportPage from utility_classes.figure_with_max_y import FigureWithMaxY, find_fig_max_y_and_generate_wrapper, extract_figure_data from cached_function_wrappers.shared import get_df_centers from cached_function_wrappers.funding_milestones_cached_functions import cached_sanitize_funding_data from cached_function_wrappers.nbs_cached_functions import cached_get_nbs_data from streamlit_constants import DASHBOARD_CONFIG_OBJECT_KEY from utility_classes.dashboard_config_parser import DashboardConfig, ExportModulePair from milestone_attribution_graph_library_module import ( make_attribution_pie, make_attribution_grouped_chart ) from constants_module import NEOSERRA_COLUMNS, OUT_COLUMNS from streamlit.delta_generator import DeltaGenerator from fiscalyear import * class CenterMilestonesReportPage(BaseReportPage): """ Implements a specific report page for visualizing center-level milestones. This class manages the lifecycle for displaying either 'Capital Funding' or 'New Business Starts' milestones. It calculates current and previous fiscal years to provide temporal context and interacts with the Neoserra export module to fetch and sanitize center-specific datasets. :param kwargs: Arbitrary keyword arguments. """ def __init__(self, **kwargs): super().__init__("Center Specific Milestones") self.fiscal_year = FiscalYear.current() self.prev_fiscal_year = self.fiscal_year.prev_fiscal_year self.fiscal_year_text = f'FY{str(self.fiscal_year.fiscal_year)[2:]}' self.prev_fiscal_year_text = f'FY{str(self.prev_fiscal_year.fiscal_year)[2:]}' self.capital_funding_report_key = "Capital Funding" self.new_business_start_report_key = "New Business Starts" def get_fiscal_year_export_url(self, selected_fiscal_year:str, is_nbs:bool) -> str: """ Determines the appropriate data export URL based on report type and year. Logic branches between New Business Start (NBS) and Funding milestones, then selects the URL corresponding to the current or previous fiscal year stored in the application configuration. :param selected_fiscal_year: The fiscal year string (e.g., 'FY26'). :type selected_fiscal_year: str :param is_nbs: Flag indicating if the report type is New Business Starts. :type is_nbs: bool :return: The URL for the requested data export. :rtype: str """ if is_nbs: export_urls:ExportModulePair = self.app_config.get_nbs_milestones_urls() else: export_urls:ExportModulePair = self.app_config.get_funding_milestones_urls() if selected_fiscal_year == self.fiscal_year_text: return export_urls.current_fy else: return export_urls.prev_fy @staticmethod def get_page_name(): """ Returns the static display name for the report page. :return: "Center Specific Milestones" :rtype: str """ return "Center Specific Milestones" def render_controls(self, container: DeltaGenerator) -> Dict[str, Any]: """ Renders UI widgets to capture reporting parameters from the user. Displays an expander containing selectors for the target fiscal year, milestone type, and specific center. It validates the connection to the data source before allowing center selection. :param container: The Streamlit container for control widgets. :type container: DeltaGenerator :return: Dictionary containing 'selected_fiscal_year', 'selected_center', and 'is_nbs'. :rtype: Dict[str, Any] """ report_settings_expander = container.expander( label="Report Options", expanded=True, key=self.get_widget_key("report_settings_expander") ) report_settings_expander.markdown("## Dataset Options") report_settings_expander.markdown("These settings will modify the input dataset used to generate the graphs.") selected_fiscal_year = report_settings_expander.selectbox( label="Fiscal Year", options=[self.prev_fiscal_year_text, self.fiscal_year_text], index=1, key=self.get_widget_key("selected_fiscal_year_selectbox") ) report_type = report_settings_expander.selectbox( label="Milestone Type", options=[self.capital_funding_report_key, self.new_business_start_report_key], index=0, key=self.get_widget_key("report_type_selectbox") ) reportable_only = report_settings_expander.checkbox(label="Reportable only?", value=True, key=self.get_widget_key("reportable_only_checkbox")) is_nbs = True if report_type == self.new_business_start_report_key else False export_url = self.get_fiscal_year_export_url(selected_fiscal_year, is_nbs) try: all_centers = get_df_centers(export_url) except Exception as e: self.logger.exception(f"Failed to fetch the dataset for this page: {e}") container.error(f"Failed to get the list of all centers for the dataset for this page. A detailed error message has been added to the logs. {self.app_config.get_errors_contact_string()}") st.stop() selected_center = report_settings_expander.selectbox( label="Selected Center", options=all_centers, index=0, key=self.get_widget_key("selected_center_selectbox") ) return { "selected_fiscal_year":selected_fiscal_year, "selected_center":selected_center, "is_nbs":is_nbs, "reportable_only":reportable_only } def generate_figures(self, parameters: Dict[str, Any]): """ Fetches raw milestone data and constructs visualization objects. Coordinates the data retrieval pipeline using cached wrappers and processes the resulting DataFrames into attribution pie charts and grouped bar charts. Figures are wrapped in utility classes to preserve metadata like Y-axis maximums for potential synchronization. :param parameters: User inputs containing center, year, and report type. :type parameters: Dict[str, Any] :return: Dictionary containing 'pie_fig', 'bar_fig', and 'milestone_df'. :rtype: Dict[str, Any] """ selected_fiscal_year = parameters['selected_fiscal_year'] selected_center = parameters['selected_center'] is_nbs = parameters['is_nbs'] reportable_only = parameters['reportable_only'] try: export_url:str = self.get_fiscal_year_export_url(selected_fiscal_year, is_nbs) if is_nbs: milestone_df = cached_get_nbs_data(export_url, reportable_only=reportable_only, allowed_centers=[selected_center]) else: milestone_df = cached_sanitize_funding_data(export_url, reportable_only=reportable_only, allowed_centers=[selected_center]) except Exception as e: self.logger.exception(f"Failed to fetch the dataset for this page: {e}") st.error( f"An issue was encountered while fetching the data for this page. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}") st.stop() title_tag = "New Business Start" if is_nbs else "Funding" try: pie_fig = make_attribution_pie( milestone_df, title=f"{selected_center} Documented vs. Not Documented {title_tag} Milestones {selected_fiscal_year}", date_note=datetime.datetime.now().strftime('%m/%d/%Y'), col_documentation_level=OUT_COLUMNS.milestone_documentation_level ) except Exception as e: self.logger.exception(f"Failed to generate the pie chart graphic with error: {e}") st.error(f"An issue was encountered while generating the pie chart graphic. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}") st.stop() try: bar_fig = make_attribution_grouped_chart( milestone_df, title=f"{selected_center} Attribution Source vs. Documentation Level For {title_tag} Milestones {selected_fiscal_year}", ) bar_fig.update_layout(height=600) except Exception as e: self.logger.exception(f"Failed to generate the bar chart graphic with error: {e}") st.error( f"An issue was encountered while generating the bar chart graphic. A detailed error message has been recorded in the logs. {self.app_config.get_errors_contact_string()}") st.stop() bar_fig_data = extract_figure_data(bar_fig) if bar_fig_data.empty: bar_fig_max_y = 0.0 else: bar_fig_max_y = bar_fig_data.groupby('x')['y'].sum().max() return { 'pie_fig': FigureWithMaxY(figure=pie_fig, max_y=0.0), 'bar_fig': FigureWithMaxY(figure=bar_fig, max_y=bar_fig_max_y), 'milestone_df':milestone_df } def render_figures(self, container: DeltaGenerator, output_data: Dict[str, Any]): """ Displays the generated visualizations and raw data in the UI. Renders Plotly charts for documentation levels and attribution sources, followed by an expander containing the underlying pandas DataFrame for auditing. :param container: The Streamlit container for output display. :type container: DeltaGenerator :param output_data: The result set from generate_figures. :type output_data: Dict[str, Any] """ milestone_df:pd.DataFrame = output_data['milestone_df'] pie_fig = output_data['pie_fig'].get("figure") bar_fig = output_data['bar_fig'].get("figure") container.plotly_chart(pie_fig, key=self.get_widget_key("funding_milestone_pie_fig")) container.plotly_chart(bar_fig, key=self.get_widget_key("funding_milestone_bar_fig")) dataset_expander = container.expander( label="Source Datasets", expanded=True, key=self.get_widget_key("dataset_expander") ) dataset_expander.markdown("## Source Data") dataset_expander.markdown("### Neoserra Milestone Dataset") dataset_expander.write(milestone_df) def get_syncable_figure_keys(self) -> List[str]: """ Designates specific figures for axis synchronization. Identifies the bar chart as a candidate for Y-axis scaling normalization when multiple instances of this report are viewed in a side-by-side comparison. :return: List of keys ('bar_fig') available for synchronization. :rtype: List[str] """ return ["bar_fig"]